chore: remove unnecessary missing-typed-dict-key suppression

Co-authored-by: Cursor <cursoragent@cursor.com>

libs/checkpoint-conformance/README.md

@@ -63,6 +63,9 @@ The suite tests **base** capabilities (required) and **extended** capabilities (
 | `delete_for_runs` | no | `adelete_for_runs` |
 | `copy_thread` | no | `acopy_thread` |
 | `prune` | no | `aprune` |
+| `delta_channel_history` | no | `aget_delta_channel_history` |
+| `delta_channel_keepset` | no | `aget_delta_channel_keepset` |
+| `delta_channel_reconstruction` | no | `aput` |
 
 Extended capabilities are detected by checking whether the method is overridden from `BaseCheckpointSaver`. If not overridden, those tests are skipped.
 

libs/checkpoint-conformance/langgraph/checkpoint/conformance/capabilities.py

@@ -23,6 +23,9 @@ class Capability(str, Enum):
     DELETE_FOR_RUNS = "delete_for_runs"
     COPY_THREAD = "copy_thread"
     PRUNE = "prune"
+    DELTA_CHANNEL_HISTORY = "delta_channel_history"
+    DELTA_CHANNEL_KEEPSET = "delta_channel_keepset"
+    DELTA_CHANNEL_RECONSTRUCTION = "delta_channel_reconstruction"
 
 
 # Capabilities that every checkpointer must support.
@@ -42,6 +45,9 @@ EXTENDED_CAPABILITIES = frozenset(
         Capability.DELETE_FOR_RUNS,
         Capability.COPY_THREAD,
         Capability.PRUNE,
+        Capability.DELTA_CHANNEL_HISTORY,
+        Capability.DELTA_CHANNEL_KEEPSET,
+        Capability.DELTA_CHANNEL_RECONSTRUCTION,
     }
 )
 
@@ -57,6 +63,9 @@ _CAPABILITY_METHOD_MAP: dict[Capability, str] = {
     Capability.DELETE_FOR_RUNS: "adelete_for_runs",
     Capability.COPY_THREAD: "acopy_thread",
     Capability.PRUNE: "aprune",
+    Capability.DELTA_CHANNEL_HISTORY: "aget_delta_channel_history",
+    Capability.DELTA_CHANNEL_KEEPSET: "aget_tuple",
+    Capability.DELTA_CHANNEL_RECONSTRUCTION: "aput",
 }
 
 

libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/__init__.py

@@ -9,6 +9,15 @@ from langgraph.checkpoint.conformance.spec.test_delete_for_runs import (
 from langgraph.checkpoint.conformance.spec.test_delete_thread import (
     run_delete_thread_tests,
 )
+from langgraph.checkpoint.conformance.spec.test_delta_channel_history import (
+    run_delta_channel_history_tests,
+)
+from langgraph.checkpoint.conformance.spec.test_delta_channel_keepset import (
+    run_delta_channel_keepset_tests,
+)
+from langgraph.checkpoint.conformance.spec.test_delta_channel_reconstruction import (
+    run_delta_channel_reconstruction_tests,
+)
 from langgraph.checkpoint.conformance.spec.test_get_tuple import run_get_tuple_tests
 from langgraph.checkpoint.conformance.spec.test_list import run_list_tests
 from langgraph.checkpoint.conformance.spec.test_prune import run_prune_tests
@@ -24,4 +33,7 @@ __all__ = [
     "run_delete_for_runs_tests",
     "run_copy_thread_tests",
     "run_prune_tests",
+    "run_delta_channel_history_tests",
+    "run_delta_channel_keepset_tests",
+    "run_delta_channel_reconstruction_tests",
 ]

libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/_delta_fixtures.py

@@ -0,0 +1,99 @@
+"""Shared fixtures for delta-channel conformance tests.
+
+Builds a parent chain with `_DeltaSnapshot` blobs at known positions via
+direct `aput` / `aput_writes` calls. No langgraph or Pregel dependency.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import Any
+from uuid import uuid4
+
+from langchain_core.runnables import RunnableConfig
+from langgraph.checkpoint.base import BaseCheckpointSaver, Checkpoint
+from langgraph.checkpoint.base.id import uuid6
+
+from langgraph.checkpoint.conformance.test_utils import generate_metadata
+
+
+async def build_delta_chain(
+    saver: BaseCheckpointSaver,
+    *,
+    thread_id: str | None = None,
+    checkpoint_ns: str = "",
+    channel: str = "messages",
+    snapshots_at_steps: Sequence[int] = (0,),
+    total_steps: int = 6,
+    write_value_fn: Any | None = None,
+) -> list[RunnableConfig]:
+    """Build a parent chain with `_DeltaSnapshot` at known positions.
+
+    Args:
+        saver: Checkpointer instance.
+        thread_id: Defaults to a random UUID.
+        checkpoint_ns: Namespace (default root).
+        channel: Channel name used for snapshots and writes.
+        snapshots_at_steps: Steps at which a `_DeltaSnapshot` blob is stored
+            in `channel_values[channel]`. Step 0 is the oldest checkpoint.
+        total_steps: Number of checkpoints in the chain.
+        write_value_fn: Callable(step) -> write value. Defaults to step index.
+
+    Returns:
+        List of stored configs (oldest first), one per step.
+    """
+    if write_value_fn is None:
+
+        def write_value_fn(step: int) -> Any:
+            return step
+
+    from langgraph.checkpoint.serde.types import _DeltaSnapshot
+
+    thread_id = thread_id or str(uuid4())
+    snapshot_set = set(snapshots_at_steps)
+    stored: list[RunnableConfig] = []
+    parent_cfg: RunnableConfig | None = None
+
+    for step in range(total_steps):
+        config: RunnableConfig = {
+            "configurable": {
+                "thread_id": thread_id,
+                "checkpoint_ns": checkpoint_ns,
+            }
+        }
+        if parent_cfg:
+            config["configurable"]["checkpoint_id"] = parent_cfg["configurable"][
+                "checkpoint_id"
+            ]
+
+        channel_values: dict[str, Any] = {}
+        channel_versions: dict[str, int] = {}
+        if step in snapshot_set:
+            channel_values[channel] = _DeltaSnapshot(
+                write_value_fn(step),
+            )
+            channel_versions[channel] = step + 1
+
+        cp = Checkpoint(
+            v=1,
+            id=str(uuid6(clock_seq=-1)),
+            ts="",
+            channel_values=channel_values,
+            channel_versions=channel_versions,
+            versions_seen={},
+            updated_channels=None,
+        )
+        new_versions = dict(channel_versions)
+        parent_cfg = await saver.aput(
+            config, cp, generate_metadata(step=step), new_versions
+        )
+        stored.append(parent_cfg)
+
+        # Write a pending write for non-snapshot steps so the walk has
+        # something to collect.
+        if step not in snapshot_set:
+            await saver.aput_writes(
+                parent_cfg, [(channel, write_value_fn(step))], str(uuid4())
+            )
+
+    return stored

libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/test_delta_channel_history.py

@@ -0,0 +1,247 @@
+"""DELTA_CHANNEL_HISTORY capability tests — aget_delta_channel_history contract."""
+
+from __future__ import annotations
+
+import traceback
+from collections.abc import Callable
+from uuid import uuid4
+
+from langgraph.checkpoint.base import BaseCheckpointSaver
+
+from langgraph.checkpoint.conformance.spec._delta_fixtures import build_delta_chain
+
+
+async def test_history_returns_writes_oldest_first(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Writes are returned oldest-to-newest."""
+    tid = str(uuid4())
+    # 5 steps: snapshot at 0, writes at 1,2,3,4.
+    # Head is step 4. Walk starts at step 3 (parent of head).
+    # Collects writes from steps 1,2,3 (between snapshot at 0 and head's parent).
+    configs = await build_delta_chain(
+        saver, thread_id=tid, channel="ch", snapshots_at_steps=[0], total_steps=5
+    )
+    head = configs[-1]
+    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
+    writes = result["ch"]["writes"]
+    values = [w[2] for w in writes]
+    assert values == [1, 2, 3], f"Expected [1,2,3], got {values}"
+
+
+async def test_history_seed_is_nearest_snapshot(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Seed is the value from the nearest ancestor with channel_values populated."""
+    tid = str(uuid4())
+    # 6 steps: snapshots at 0 and 3, writes at 1,2,4,5.
+    # Head is step 5. Walk from step 4 backward stops at step 3 (snapshot).
+    # Collects writes from step 4 only (between step 3 and head's parent step 4).
+    configs = await build_delta_chain(
+        saver,
+        thread_id=tid,
+        channel="ch",
+        snapshots_at_steps=[0, 3],
+        total_steps=6,
+    )
+    head = configs[-1]
+    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
+    assert "seed" in result["ch"], "Expected seed from snapshot at step 3"
+    seed = result["ch"]["seed"]
+    from langgraph.checkpoint.serde.types import _DeltaSnapshot
+
+    actual_value = seed.value if isinstance(seed, _DeltaSnapshot) else seed
+    assert actual_value == 3, f"Expected seed value 3 (step 3), got {actual_value}"
+    writes = result["ch"]["writes"]
+    values = [w[2] for w in writes]
+    assert values == [4], f"Expected [4], got {values}"
+
+
+async def test_history_excludes_target_pending_writes(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Target's own pending_writes are NOT included in the history."""
+    tid = str(uuid4())
+    configs = await build_delta_chain(
+        saver, thread_id=tid, channel="ch", snapshots_at_steps=[0], total_steps=3
+    )
+    head = configs[-1]
+    # Add writes directly to the head checkpoint
+    await saver.aput_writes(head, [("ch", "extra")], str(uuid4()))
+    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
+    writes = result["ch"]["writes"]
+    values = [w[2] for w in writes]
+    assert "extra" not in values, f"Target's writes should be excluded, got {values}"
+
+
+async def test_history_multi_channel(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Multiple channels have independent walk termination."""
+    tid = str(uuid4())
+    configs: list = []
+    parent_cfg = None
+    from langgraph.checkpoint.base import Checkpoint
+    from langgraph.checkpoint.base.id import uuid6
+    from langgraph.checkpoint.serde.types import _DeltaSnapshot
+
+    from langgraph.checkpoint.conformance.test_utils import generate_metadata
+
+    for step in range(5):
+        config = {"configurable": {"thread_id": tid, "checkpoint_ns": ""}}
+        if parent_cfg:
+            config["configurable"]["checkpoint_id"] = parent_cfg["configurable"][
+                "checkpoint_id"
+            ]
+        cv: dict = {}
+        cvs: dict = {}
+        if step == 1:
+            cv["a"] = _DeltaSnapshot("snap_a")
+            cvs["a"] = step + 1
+        if step == 3:
+            cv["b"] = _DeltaSnapshot("snap_b")
+            cvs["b"] = step + 1
+        cp = Checkpoint(
+            v=1,
+            id=str(uuid6(clock_seq=-1)),
+            ts="",
+            channel_values=cv,
+            channel_versions=cvs,
+            versions_seen={},
+            updated_channels=None,
+        )
+        parent_cfg = await saver.aput(config, cp, generate_metadata(step=step), cvs)
+        configs.append(parent_cfg)
+        await saver.aput_writes(parent_cfg, [("a", step), ("b", step)], str(uuid4()))
+
+    head = configs[-1]
+    result = await saver.aget_delta_channel_history(config=head, channels=["a", "b"])
+    a_writes = [w[2] for w in result["a"]["writes"]]
+    b_writes = [w[2] for w in result["b"]["writes"]]
+    assert a_writes == [1, 2, 3], f"Expected a writes [1,2,3], got {a_writes}"
+    assert b_writes == [3], f"Expected b writes [3], got {b_writes}"
+
+
+async def test_history_empty_channels_returns_empty(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Empty channels list returns empty mapping."""
+    tid = str(uuid4())
+    configs = await build_delta_chain(
+        saver, thread_id=tid, channel="ch", snapshots_at_steps=[0], total_steps=3
+    )
+    result = await saver.aget_delta_channel_history(config=configs[-1], channels=[])
+    assert result == {}
+
+
+async def test_history_walk_to_root_no_seed(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Walk reaches root without finding seed — no 'seed' key in result."""
+    tid = str(uuid4())
+    configs = await build_delta_chain(
+        saver,
+        thread_id=tid,
+        channel="ch",
+        snapshots_at_steps=[],
+        total_steps=4,
+    )
+    head = configs[-1]
+    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
+    assert "seed" not in result["ch"], f"Expected no seed, got {result['ch']}"
+
+
+async def test_history_migration_plain_value_as_seed(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Pre-delta plain value in channel_values acts as seed (migration case).
+
+    When a thread was originally using a regular channel (BinaryOperatorAggregate)
+    and later switches to DeltaChannel, the old checkpoint has a plain value in
+    channel_values[ch] (not a _DeltaSnapshot). The walk should treat it as the
+    seed and terminate there.
+    """
+    from langgraph.checkpoint.base import Checkpoint
+    from langgraph.checkpoint.base.id import uuid6
+
+    from langgraph.checkpoint.conformance.test_utils import generate_metadata
+
+    tid = str(uuid4())
+    configs: list = []
+    parent_cfg = None
+
+    for step in range(4):
+        config = {"configurable": {"thread_id": tid, "checkpoint_ns": ""}}
+        if parent_cfg:
+            config["configurable"]["checkpoint_id"] = parent_cfg["configurable"][
+                "checkpoint_id"
+            ]
+        cv: dict = {}
+        cvs: dict = {}
+        # Step 1: plain value (migration case — old checkpoint before delta)
+        if step == 1:
+            cv["ch"] = [10, 20, 30]
+            cvs["ch"] = step + 1
+        cp = Checkpoint(
+            v=1,
+            id=str(uuid6(clock_seq=-1)),
+            ts="",
+            channel_values=cv,
+            channel_versions=cvs,
+            versions_seen={},
+            updated_channels=None,
+        )
+        parent_cfg = await saver.aput(config, cp, generate_metadata(step=step), cvs)
+        configs.append(parent_cfg)
+        if step != 1:
+            await saver.aput_writes(parent_cfg, [("ch", step)], str(uuid4()))
+
+    head = configs[-1]
+    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
+    # Seed should be the plain value from step 1
+    assert "seed" in result["ch"], "Expected seed from migration plain value at step 1"
+    seed = result["ch"]["seed"]
+    assert seed == [10, 20, 30], f"Expected plain value [10,20,30], got {seed}"
+    # Writes should be from step 2 only (between seed at step 1 and head's parent step 2)
+    writes = result["ch"]["writes"]
+    values = [w[2] for w in writes]
+    assert values == [2], f"Expected [2], got {values}"
+
+
+ALL_DELTA_CHANNEL_HISTORY_TESTS = [
+    test_history_returns_writes_oldest_first,
+    test_history_seed_is_nearest_snapshot,
+    test_history_excludes_target_pending_writes,
+    test_history_multi_channel,
+    test_history_empty_channels_returns_empty,
+    test_history_walk_to_root_no_seed,
+    test_history_migration_plain_value_as_seed,
+]
+
+
+async def run_delta_channel_history_tests(
+    saver: BaseCheckpointSaver,
+    on_test_result: Callable[[str, str, bool, str | None], None] | None = None,
+) -> tuple[int, int, list[str]]:
+    """Run all delta_channel_history tests. Returns (passed, failed, failure_names)."""
+    passed = 0
+    failed = 0
+    failures: list[str] = []
+    for test_fn in ALL_DELTA_CHANNEL_HISTORY_TESTS:
+        try:
+            await test_fn(saver)
+            passed += 1
+            if on_test_result:
+                on_test_result("delta_channel_history", test_fn.__name__, True, None)
+        except Exception:
+            failed += 1
+            msg = f"{test_fn.__name__}: {traceback.format_exc()}"
+            failures.append(msg)
+            if on_test_result:
+                on_test_result(
+                    "delta_channel_history",
+                    test_fn.__name__,
+                    False,
+                    traceback.format_exc(),
+                )
+    return passed, failed, failures

libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/test_delta_channel_keepset.py

@@ -0,0 +1,172 @@
+"""DELTA_CHANNEL_KEEPSET capability tests — aget_delta_channel_keepset contract."""
+
+from __future__ import annotations
+
+import traceback
+from collections.abc import Callable
+from uuid import uuid4
+
+from langgraph.checkpoint.base import BaseCheckpointSaver
+
+from langgraph.checkpoint.conformance.spec._delta_fixtures import build_delta_chain
+
+
+async def test_keepset_empty_channels_returns_target_only(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Empty channels → keep-set is just {target_id}."""
+    tid = str(uuid4())
+    configs = await build_delta_chain(
+        saver, thread_id=tid, channel="ch", snapshots_at_steps=[0], total_steps=4
+    )
+    head = configs[-1]
+    keep = await saver.aget_delta_channel_keepset(config=head, channels=[])
+    head_id = head["configurable"]["checkpoint_id"]
+    assert keep == {head_id}, f"Expected only target, got {keep}"
+
+
+async def test_keepset_snapshot_at_target(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """When target itself has a snapshot, keep-set is just {target_id}."""
+    tid = str(uuid4())
+    configs = await build_delta_chain(
+        saver,
+        thread_id=tid,
+        channel="ch",
+        snapshots_at_steps=[0, 3],
+        total_steps=4,
+    )
+    head = configs[3]
+    keep = await saver.aget_delta_channel_keepset(config=head, channels=["ch"])
+    head_id = head["configurable"]["checkpoint_id"]
+    assert keep == {head_id}, f"Snapshot at target should yield only target, got {keep}"
+
+
+async def test_keepset_snapshot_n_back(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Snapshot N steps back → target + intermediates + snapshot ancestor."""
+    tid = str(uuid4())
+    configs = await build_delta_chain(
+        saver, thread_id=tid, channel="ch", snapshots_at_steps=[0], total_steps=5
+    )
+    head = configs[-1]
+    keep = await saver.aget_delta_channel_keepset(config=head, channels=["ch"])
+    expected_ids = {c["configurable"]["checkpoint_id"] for c in configs}
+    assert keep == expected_ids, f"Expected all ancestors, got {keep}"
+
+
+async def test_keepset_multi_channel_union(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Multi-channel keep-set is the union (max chain per channel)."""
+    tid = str(uuid4())
+    from langgraph.checkpoint.base import Checkpoint
+    from langgraph.checkpoint.base.id import uuid6
+    from langgraph.checkpoint.serde.types import _DeltaSnapshot
+
+    from langgraph.checkpoint.conformance.test_utils import generate_metadata
+
+    configs: list = []
+    parent_cfg = None
+    for step in range(6):
+        config = {"configurable": {"thread_id": tid, "checkpoint_ns": ""}}
+        if parent_cfg:
+            config["configurable"]["checkpoint_id"] = parent_cfg["configurable"][
+                "checkpoint_id"
+            ]
+        cv: dict = {}
+        cvs: dict = {}
+        # Channel "a" has snapshot at step 3 (recent)
+        if step == 3:
+            cv["a"] = _DeltaSnapshot("snap_a")
+            cvs["a"] = step + 1
+        # Channel "b" has snapshot at step 1 (further back)
+        if step == 1:
+            cv["b"] = _DeltaSnapshot("snap_b")
+            cvs["b"] = step + 1
+        cp = Checkpoint(
+            v=1,
+            id=str(uuid6(clock_seq=-1)),
+            ts="",
+            channel_values=cv,
+            channel_versions=cvs,
+            versions_seen={},
+            updated_channels=None,
+        )
+        parent_cfg = await saver.aput(config, cp, generate_metadata(step=step), cvs)
+        configs.append(parent_cfg)
+
+    head = configs[-1]
+    keep = await saver.aget_delta_channel_keepset(config=head, channels=["a", "b"])
+    # Union: b needs back to step 1, so steps 1..5 (all except step 0) plus head
+    expected_ids = {c["configurable"]["checkpoint_id"] for c in configs[1:]}
+    expected_ids.add(head["configurable"]["checkpoint_id"])
+    assert keep == expected_ids, f"Expected union, got {keep} vs {expected_ids}"
+
+
+async def test_keepset_walk_to_root(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """No snapshot anywhere → entire chain to root is in keep-set."""
+    tid = str(uuid4())
+    configs = await build_delta_chain(
+        saver, thread_id=tid, channel="ch", snapshots_at_steps=[], total_steps=4
+    )
+    head = configs[-1]
+    keep = await saver.aget_delta_channel_keepset(config=head, channels=["ch"])
+    all_ids = {c["configurable"]["checkpoint_id"] for c in configs}
+    assert keep == all_ids, f"Expected full chain, got {keep}"
+
+
+async def test_keepset_deterministic(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Same inputs return identical sets."""
+    tid = str(uuid4())
+    configs = await build_delta_chain(
+        saver, thread_id=tid, channel="ch", snapshots_at_steps=[0], total_steps=5
+    )
+    head = configs[-1]
+    keep1 = await saver.aget_delta_channel_keepset(config=head, channels=["ch"])
+    keep2 = await saver.aget_delta_channel_keepset(config=head, channels=["ch"])
+    assert keep1 == keep2
+
+
+ALL_DELTA_CHANNEL_KEEPSET_TESTS = [
+    test_keepset_empty_channels_returns_target_only,
+    test_keepset_snapshot_at_target,
+    test_keepset_snapshot_n_back,
+    test_keepset_multi_channel_union,
+    test_keepset_walk_to_root,
+    test_keepset_deterministic,
+]
+
+
+async def run_delta_channel_keepset_tests(
+    saver: BaseCheckpointSaver,
+    on_test_result: Callable[[str, str, bool, str | None], None] | None = None,
+) -> tuple[int, int, list[str]]:
+    """Run all delta_channel_keepset tests. Returns (passed, failed, failure_names)."""
+    passed = 0
+    failed = 0
+    failures: list[str] = []
+    for test_fn in ALL_DELTA_CHANNEL_KEEPSET_TESTS:
+        try:
+            await test_fn(saver)
+            passed += 1
+            if on_test_result:
+                on_test_result("delta_channel_keepset", test_fn.__name__, True, None)
+        except Exception:
+            failed += 1
+            msg = f"{test_fn.__name__}: {traceback.format_exc()}"
+            failures.append(msg)
+            if on_test_result:
+                on_test_result(
+                    "delta_channel_keepset",
+                    test_fn.__name__,
+                    False,
+                    traceback.format_exc(),
+                )
+    return passed, failed, failures

libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/test_delta_channel_reconstruction.py

@@ -0,0 +1,164 @@
+"""DELTA_CHANNEL_RECONSTRUCTION capability tests — end-to-end round-trip.
+
+Exercises: aput + aput_writes + aget_delta_channel_history + reconstruction.
+This catches the most common silent-corruption mode: failing to round-trip
+`_DeltaSnapshot` blobs through serialization.
+
+NOTE: This test does NOT import from `langgraph` (which is not a dependency
+of checkpoint-conformance). Instead it inlines a minimal reconstruction
+equivalent: seed + fold writes through a simple list-append reducer.
+"""
+
+from __future__ import annotations
+
+import traceback
+from collections.abc import Callable
+from typing import Any
+from uuid import uuid4
+
+from langgraph.checkpoint.base import BaseCheckpointSaver
+
+from langgraph.checkpoint.conformance.spec._delta_fixtures import build_delta_chain
+
+
+def _reconstruct(seed: Any, writes: list) -> list:
+    """Minimal DeltaChannel reconstruction: list-append reducer.
+
+    Mirrors DeltaChannel.from_checkpoint(seed) + replay_writes(writes).
+    """
+    from langgraph.checkpoint.serde.types import _DeltaSnapshot
+
+    if seed is None:
+        base: list = []
+    elif isinstance(seed, _DeltaSnapshot):
+        base = list(seed.value)
+    else:
+        base = list(seed)
+    for _task_id, _ch, value in writes:
+        base = base + value
+    return base
+
+
+async def test_reconstruction_basic(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Reconstruct delta channel value from history matches expected."""
+    tid = str(uuid4())
+    # 5 steps: snapshot at 0 (value=[0]), writes at 1,2,3,4.
+    # Head = step 4. Walk from parent (step 3) collects writes 1,2,3.
+    configs = await build_delta_chain(
+        saver,
+        thread_id=tid,
+        channel="msgs",
+        snapshots_at_steps=[0],
+        total_steps=5,
+        write_value_fn=lambda step: [step],
+    )
+    head = configs[-1]
+    result = await saver.aget_delta_channel_history(config=head, channels=["msgs"])
+    history = result["msgs"]
+
+    seed = history.get("seed")
+    reconstructed = _reconstruct(seed, history["writes"])
+
+    # seed=[0] from step 0, writes from steps 1,2,3
+    expected = [0] + [1] + [2] + [3]
+    assert reconstructed == expected, (
+        f"Reconstructed {reconstructed} != expected {expected}"
+    )
+
+
+async def test_reconstruction_mid_chain_snapshot(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Reconstruction works when snapshot is mid-chain."""
+    tid = str(uuid4())
+    # 6 steps: snapshots at 0 and 3, writes at 1,2,4,5.
+    # Head = step 5. Walk from step 4 stops at step 3 (snapshot).
+    # Collects writes from step 4.
+    configs = await build_delta_chain(
+        saver,
+        thread_id=tid,
+        channel="msgs",
+        snapshots_at_steps=[0, 3],
+        total_steps=6,
+        write_value_fn=lambda step: [step],
+    )
+    head = configs[-1]
+    result = await saver.aget_delta_channel_history(config=head, channels=["msgs"])
+    history = result["msgs"]
+
+    seed = history.get("seed")
+    reconstructed = _reconstruct(seed, history["writes"])
+
+    # Snapshot at step 3 = [3], write from step 4
+    expected = [3] + [4]
+    assert reconstructed == expected, (
+        f"Reconstructed {reconstructed} != expected {expected}"
+    )
+
+
+async def test_reconstruction_no_snapshot(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Reconstruction from root (no snapshot) gives all writes accumulated."""
+    tid = str(uuid4())
+    # 4 steps: no snapshot, writes at 0,1,2,3.
+    # Head = step 3. Walk from step 2 collects writes 0,1,2.
+    configs = await build_delta_chain(
+        saver,
+        thread_id=tid,
+        channel="msgs",
+        snapshots_at_steps=[],
+        total_steps=4,
+        write_value_fn=lambda step: [step],
+    )
+    head = configs[-1]
+    result = await saver.aget_delta_channel_history(config=head, channels=["msgs"])
+    history = result["msgs"]
+
+    seed = history.get("seed")
+    reconstructed = _reconstruct(seed, history["writes"])
+
+    # No seed → start empty, writes from steps 0,1,2
+    expected = [0] + [1] + [2]
+    assert reconstructed == expected, (
+        f"Reconstructed {reconstructed} != expected {expected}"
+    )
+
+
+ALL_DELTA_CHANNEL_RECONSTRUCTION_TESTS = [
+    test_reconstruction_basic,
+    test_reconstruction_mid_chain_snapshot,
+    test_reconstruction_no_snapshot,
+]
+
+
+async def run_delta_channel_reconstruction_tests(
+    saver: BaseCheckpointSaver,
+    on_test_result: Callable[[str, str, bool, str | None], None] | None = None,
+) -> tuple[int, int, list[str]]:
+    """Run all reconstruction tests. Returns (passed, failed, failure_names)."""
+    passed = 0
+    failed = 0
+    failures: list[str] = []
+    for test_fn in ALL_DELTA_CHANNEL_RECONSTRUCTION_TESTS:
+        try:
+            await test_fn(saver)
+            passed += 1
+            if on_test_result:
+                on_test_result(
+                    "delta_channel_reconstruction", test_fn.__name__, True, None
+                )
+        except Exception:
+            failed += 1
+            msg = f"{test_fn.__name__}: {traceback.format_exc()}"
+            failures.append(msg)
+            if on_test_result:
+                on_test_result(
+                    "delta_channel_reconstruction",
+                    test_fn.__name__,
+                    False,
+                    traceback.format_exc(),
+                )
+    return passed, failed, failures

libs/checkpoint-conformance/langgraph/checkpoint/conformance/validate.py

@@ -19,6 +19,15 @@ from langgraph.checkpoint.conformance.spec.test_delete_for_runs import (
 from langgraph.checkpoint.conformance.spec.test_delete_thread import (
     run_delete_thread_tests,
 )
+from langgraph.checkpoint.conformance.spec.test_delta_channel_history import (
+    run_delta_channel_history_tests,
+)
+from langgraph.checkpoint.conformance.spec.test_delta_channel_keepset import (
+    run_delta_channel_keepset_tests,
+)
+from langgraph.checkpoint.conformance.spec.test_delta_channel_reconstruction import (
+    run_delta_channel_reconstruction_tests,
+)
 from langgraph.checkpoint.conformance.spec.test_get_tuple import run_get_tuple_tests
 from langgraph.checkpoint.conformance.spec.test_list import run_list_tests
 from langgraph.checkpoint.conformance.spec.test_prune import run_prune_tests
@@ -35,6 +44,9 @@ _RUNNERS = {
     Capability.DELETE_FOR_RUNS: run_delete_for_runs_tests,
     Capability.COPY_THREAD: run_copy_thread_tests,
     Capability.PRUNE: run_prune_tests,
+    Capability.DELTA_CHANNEL_HISTORY: run_delta_channel_history_tests,
+    Capability.DELTA_CHANNEL_KEEPSET: run_delta_channel_keepset_tests,
+    Capability.DELTA_CHANNEL_RECONSTRUCTION: run_delta_channel_reconstruction_tests,
 }
 
 

libs/checkpoint-conformance/pyproject.toml

@@ -43,7 +43,11 @@ asyncio_mode = "auto"
 # The extended methods (acopy_thread, adelete_for_runs, aprune) are checked
 # at runtime via capability detection and may not exist on the installed
 # base class. Dict literal inference is also overly strict for RunnableConfig.
+# Delta-channel tests import from `langgraph` (not a declared dep of this
+# package — at test time it is installed alongside); private `_DeltaSnapshot`
+# imports are intentional (beta surface).
 unresolved-attribute = "ignore"
+unresolved-import = "ignore"
 invalid-argument-type = "ignore"
 invalid-return-type = "ignore"
 
@@ -58,6 +62,9 @@ lint.select = [
 lint.ignore = ["E501", "B008"]
 target-version = "py310"
 
+[tool.uv.sources]
+langgraph-checkpoint = {path = "../checkpoint", editable = true}
+
 [[tool.uv.index]]
 name = "testpypi"
 url = "https://test.pypi.org/simple/"

libs/checkpoint-sqlite/tests/test_conformance_delta.py

@@ -0,0 +1,37 @@
+"""Run delta-channel conformance capabilities against AsyncSqliteSaver."""
+
+from __future__ import annotations
+
+import pytest
+
+pytest.importorskip(
+    "langgraph.checkpoint.conformance",
+    reason="langgraph-checkpoint-conformance not installed",
+)
+pytest.importorskip("aiosqlite", reason="aiosqlite not installed")
+
+
+@pytest.mark.asyncio
+async def test_delta_channel_conformance():
+    from langgraph.checkpoint.conformance import validate
+    from langgraph.checkpoint.conformance.initializer import checkpointer_test
+
+    from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
+
+    @checkpointer_test(name="AsyncSqliteSaver")
+    async def sqlite_saver():
+        async with AsyncSqliteSaver.from_conn_string(":memory:") as saver:
+            yield saver
+
+    report = await validate(
+        sqlite_saver,
+        capabilities={
+            "delta_channel_history",
+            "delta_channel_keepset",
+            "delta_channel_reconstruction",
+        },
+    )
+    for cap, result in report.results.items():
+        if result.passed is False:
+            details = "\n".join(result.failures or [])
+            pytest.fail(f"Capability {cap} failed:\n{details}")

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

@@ -327,6 +327,14 @@ class BaseCheckpointSaver(Generic[V]):
 
         Args:
             run_ids: The run IDs whose checkpoints should be deleted.
+
+        !!! warning "DeltaChannel"
+
+            Deleting a run that produced ancestor `checkpoint_writes` — or
+            the only `_DeltaSnapshot` blob — for a still-live thread will
+            break reconstruction of any `DeltaChannel` whose history
+            depended on those rows. See the `DeltaChannel` note on `prune`
+            for safe-recovery strategies.
         """
         raise NotImplementedError
 
@@ -340,6 +348,17 @@ class BaseCheckpointSaver(Generic[V]):
         Args:
             source_thread_id: The thread ID to copy from.
             target_thread_id: The thread ID to copy to.
+
+        !!! warning "DeltaChannel"
+
+            Implementations must copy the **complete** parent chain (all
+            ancestor checkpoints and their `checkpoint_writes`) — copying
+            only the head checkpoint will leave the target thread with
+            `DeltaChannel` state that cannot be reconstructed (no path back
+            to a `_DeltaSnapshot` ancestor). Equivalently, the copy must
+            include enough ancestors that every `DeltaChannel`-backed key
+            has either a `_DeltaSnapshot` in `channel_values` somewhere in
+            the chain, or a complete write history back to the chain root.
         """
         raise NotImplementedError
 
@@ -355,6 +374,34 @@ class BaseCheckpointSaver(Generic[V]):
             thread_ids: The thread IDs to prune.
             strategy: The pruning strategy. `"keep_latest"` retains only the most
                 recent checkpoint per namespace. `"delete"` removes all checkpoints.
+
+        !!! warning "DeltaChannel"
+
+            Custom implementations must be `DeltaChannel`-aware. `DeltaChannel`
+            stores only a sentinel in `channel_values` for non-snapshot steps;
+            reconstruction walks the parent chain via
+            `get_delta_channel_history`, accumulating rows from
+            `checkpoint_writes` until it reaches an ancestor whose
+            `channel_values` contains a `_DeltaSnapshot` blob (written every
+            `snapshot_frequency` updates).
+
+            A naive `"keep_latest"` that drops intermediate checkpoints and
+            their writes can sever that chain: the surviving "latest"
+            checkpoint is rarely a snapshot point itself, so its delta
+            channels would silently reconstruct as empty (no error raised —
+            `get_delta_channel_history` simply returns no `seed`). Safe
+            options when the graph uses `DeltaChannel`:
+
+            * Walk back from each kept checkpoint and preserve every
+              ancestor (plus its `checkpoint_writes`) up to the nearest one
+              whose `channel_values` already contains a `_DeltaSnapshot` for
+              every `DeltaChannel`-backed key.
+            * Force a fresh snapshot on the kept checkpoint before deleting
+              ancestors — rewrite `channel_values[k] = _DeltaSnapshot(value)`
+              for each delta channel `k` (resolving `value` via the existing
+              ancestor walk first), then prune.
+            * Skip pruning threads whose graph uses `DeltaChannel` until one
+              of the above is implemented.
         """
         raise NotImplementedError
 
@@ -471,6 +518,13 @@ class BaseCheckpointSaver(Generic[V]):
 
         Args:
             run_ids: The run IDs whose checkpoints should be deleted.
+
+        !!! warning "DeltaChannel"
+
+            See `delete_for_runs` — deleting rows a still-live thread's
+            `DeltaChannel` reconstruction depends on (writes between the
+            head and its nearest `_DeltaSnapshot` ancestor) will silently
+            corrupt that channel's state.
         """
         raise NotImplementedError
 
@@ -484,6 +538,13 @@ class BaseCheckpointSaver(Generic[V]):
         Args:
             source_thread_id: The thread ID to copy from.
             target_thread_id: The thread ID to copy to.
+
+        !!! warning "DeltaChannel"
+
+            See `copy_thread` — the copy must carry the complete parent
+            chain (or at least back to a `_DeltaSnapshot` ancestor for every
+            `DeltaChannel`) so the target thread can reconstruct delta
+            state.
         """
         raise NotImplementedError
 
@@ -499,6 +560,13 @@ class BaseCheckpointSaver(Generic[V]):
             thread_ids: The thread IDs to prune.
             strategy: The pruning strategy. `"keep_latest"` retains only the most
                 recent checkpoint per namespace. `"delete"` removes all checkpoints.
+
+        !!! warning "DeltaChannel"
+
+            See `prune` for the full `DeltaChannel` caveat. In short:
+            `"keep_latest"` must not drop ancestor checkpoints / writes that
+            sit between the kept checkpoint and the nearest `_DeltaSnapshot`
+            ancestor, or delta channels will silently reconstruct as empty.
         """
         raise NotImplementedError
 
@@ -612,6 +680,123 @@ class BaseCheckpointSaver(Generic[V]):
             result[ch] = entry
         return result
 
+    def get_delta_channel_keepset(
+        self,
+        *,
+        config: RunnableConfig,
+        channels: Sequence[str],
+    ) -> set[str]:
+        """Return ancestor checkpoint_ids that must survive deletion.
+
+        !!! warning "Beta"
+
+            This method is part of the `DeltaChannel` support surface and is
+            in beta. The signature may change while the delta-channel design
+            stabilizes.
+
+        Walks the parent chain from `config` backward, collecting visited
+        checkpoint_ids (inclusive of the target), and terminates per-channel
+        when that channel has a populated `channel_values[ch]` (a
+        `_DeltaSnapshot` blob or a pre-migration plain value). The returned
+        set is the minimum keep-set: every checkpoint_id whose removal would
+        break reconstruction of the listed channels at `config`.
+
+        Pass `channels=[]` to return just `{config.checkpoint_id}` — useful
+        for graphs that don't use `DeltaChannel`.
+
+        Compose this into custom `prune` / `delete_for_runs` / `copy_thread`::
+
+            keep = saver.get_delta_channel_keepset(
+                config=head_config, channels=delta_channels,
+            )
+            delete_rows_not_in(keep)
+
+        Note:
+            The default implementation here uses repeated `get_tuple` calls
+            to walk the parent chain. This is a basic reference implementation
+            suitable for low-frequency maintenance operations (prune, etc.).
+            Custom checkpointer backends may override with a more efficient
+            version tailored to their data model (e.g. a single SQL query
+            with a recursive CTE), but it is not required — the default works
+            correctly for any saver that implements `get_tuple`.
+
+        Args:
+            config: Configuration identifying the target checkpoint.
+            channels: Channel names whose delta history must be preserved.
+                Empty sequence means only the target checkpoint_id is kept.
+
+        Returns:
+            Set of checkpoint_ids that must not be deleted.
+        """
+        target_tuple = self.get_tuple(config)
+        if target_tuple is None:
+            return set()
+        target_id = target_tuple.config["configurable"]["checkpoint_id"]
+        keep: set[str] = {target_id}
+        if not channels:
+            return keep
+        remaining: set[str] = set(channels)
+        for ch in list(remaining):
+            if ch in target_tuple.checkpoint["channel_values"]:
+                remaining.discard(ch)
+        if not remaining:
+            return keep
+        cursor_config: RunnableConfig | None = target_tuple.parent_config
+        while cursor_config is not None and remaining:
+            tup = self.get_tuple(cursor_config)
+            if tup is None:
+                break
+            cid = tup.config["configurable"]["checkpoint_id"]
+            keep.add(cid)
+            for ch in list(remaining):
+                if ch in tup.checkpoint["channel_values"]:
+                    remaining.discard(ch)
+            if not remaining:
+                break
+            cursor_config = tup.parent_config
+        return keep
+
+    async def aget_delta_channel_keepset(
+        self,
+        *,
+        config: RunnableConfig,
+        channels: Sequence[str],
+    ) -> set[str]:
+        """Async version of `get_delta_channel_keepset`.
+
+        !!! warning "Beta"
+
+            This method is part of the `DeltaChannel` support surface and is
+            in beta. See `get_delta_channel_keepset` for full documentation.
+        """
+        target_tuple = await self.aget_tuple(config)
+        if target_tuple is None:
+            return set()
+        target_id = target_tuple.config["configurable"]["checkpoint_id"]
+        keep: set[str] = {target_id}
+        if not channels:
+            return keep
+        remaining: set[str] = set(channels)
+        for ch in list(remaining):
+            if ch in target_tuple.checkpoint["channel_values"]:
+                remaining.discard(ch)
+        if not remaining:
+            return keep
+        cursor_config: RunnableConfig | None = target_tuple.parent_config
+        while cursor_config is not None and remaining:
+            tup = await self.aget_tuple(cursor_config)
+            if tup is None:
+                break
+            cid = tup.config["configurable"]["checkpoint_id"]
+            keep.add(cid)
+            for ch in list(remaining):
+                if ch in tup.checkpoint["channel_values"]:
+                    remaining.discard(ch)
+            if not remaining:
+                break
+            cursor_config = tup.parent_config
+        return keep
+
     def get_next_version(self, current: V | None, channel: None) -> V:
         """Generate the next version ID for a channel.
 

libs/checkpoint/tests/test_conformance_delta.py

@@ -0,0 +1,35 @@
+"""Run delta-channel conformance capabilities against InMemorySaver."""
+
+from __future__ import annotations
+
+import pytest
+
+conformance = pytest.importorskip(
+    "langgraph.checkpoint.conformance",
+    reason="langgraph-checkpoint-conformance not installed",
+)
+
+
+@pytest.mark.asyncio
+async def test_delta_channel_conformance():
+    from langgraph.checkpoint.conformance import validate
+    from langgraph.checkpoint.conformance.initializer import checkpointer_test
+
+    from langgraph.checkpoint.memory import InMemorySaver
+
+    @checkpointer_test(name="InMemorySaver")
+    async def mem_saver():
+        yield InMemorySaver()
+
+    report = await validate(
+        mem_saver,
+        capabilities={
+            "delta_channel_history",
+            "delta_channel_keepset",
+            "delta_channel_reconstruction",
+        },
+    )
+    for cap, result in report.results.items():
+        if result.passed is False:
+            details = "\n".join(result.failures or [])
+            pytest.fail(f"Capability {cap} failed:\n{details}")