done

docs/delta-channel-checkpointer-guide.md

@@ -0,0 +1,402 @@
+# DeltaChannel Checkpointer Guide
+
+> **Beta** — `DeltaChannel` and all APIs described here (`_DeltaSnapshot`,
+> `get_delta_channel_history`, `get_delta_channel_keepset`,
+> `delta_updates_since_snapshot` metadata) are in beta. The on-disk
+> representation and method signatures may change in future releases.
+
+This guide is for third-party `BaseCheckpointSaver` authors who need to
+support graphs using `DeltaChannel`. It covers:
+
+1. How `DeltaChannel` state is stored across checkpoint tables.
+2. Which saver methods interact with delta channels and what the OSS
+   implementations look like (working references you can copy).
+3. How to correctly implement cleanup methods (`prune`, `delete_for_runs`,
+   `copy_thread`) without silently corrupting delta history.
+
+---
+
+## Part I — Background
+
+### How DeltaChannel is stored
+
+Unlike regular channels, `DeltaChannel` does **not** write its accumulated
+value into every checkpoint. Instead:
+
+- `DeltaChannel.checkpoint()` always returns `MISSING`
+  (`libs/langgraph/langgraph/channels/delta.py`). The channel's live value
+  is never placed in `channel_values` by default.
+
+- A full **snapshot** (`_DeltaSnapshot(value)`) is written into
+  `channel_values[k]` only when the channel's update count reaches
+  `snapshot_frequency` (default 1000). This decision is made by
+  `delta_channels_to_snapshot()` in `libs/langgraph/langgraph/pregel/_checkpoint.py`.
+
+- Between snapshots, the channel's writes are stored as ordinary
+  `checkpoint_writes` rows (Postgres) or `writes` rows (SQLite), keyed by
+  `(thread_id, checkpoint_ns, checkpoint_id, task_id, idx)`.
+
+- Reconstruction at read time walks the parent chain backward from the
+  target checkpoint, collecting writes, until it finds an ancestor whose
+  `channel_values[ch]` is populated (the "seed"). It then replays all
+  collected writes through the channel's reducer on top of that seed.
+
+#### Storage layout per backend
+
+| Backend | Snapshot blob | Writes | Key structure |
+|---------|--------------|--------|---------------|
+| **Postgres** | `checkpoint_blobs` table; checkpoint JSON has `True` sentinel | `checkpoint_writes` table | `(thread_id, checkpoint_ns, channel, version)` for blobs |
+| **SQLite** | Inline in checkpoint blob | `writes` table | `(thread_id, checkpoint_ns, checkpoint_id, task_id, idx)` |
+| **InMemory** | `self.blobs` dict | `self.writes` dict | Same logical key structure |
+
+#### The metadata hint
+
+`CheckpointMetadata.delta_updates_since_snapshot` (`dict[str, int]`) tracks
+how many supersteps have written to each delta channel since the last
+snapshot. It resets to 0 when a snapshot fires. This is a **hint** for the
+pregel loop — the source of truth for "is there a snapshot here?" is whether
+`channel_values[ch]` is populated in the checkpoint.
+
+### The reconstruction walk
+
+```mermaid
+flowchart LR
+    subgraph parentChain [Parent Chain]
+        direction RL
+        HEAD["HEAD checkpoint<br/>(target)"]
+        P1["Parent 1<br/>writes: [w4]"]
+        P2["Parent 2<br/>writes: [w3]"]
+        P3["Parent 3<br/>_DeltaSnapshot(seed)<br/>writes: [w2]"]
+        P4["Parent 4<br/>writes: [w1]"]
+    end
+    HEAD --> P1
+    P1 --> P2
+    P2 --> P3
+    P3 --> P4
+
+    subgraph result [History result]
+        SEED["seed = _DeltaSnapshot(...)"]
+        WRITES["writes = [w3, w4]<br/>(oldest → newest)"]
+    end
+    P3 -.-> SEED
+    P2 -.-> WRITES
+    P1 -.-> WRITES
+```
+
+The walk:
+1. Starts at `target.parent_config` (target's own `pending_writes` are
+   excluded — they belong to the *next* super-step).
+2. At each ancestor, collects `pending_writes` for the requested channels.
+3. Stops per-channel when `channel_values[ch]` is non-empty (= seed found).
+4. Returns `{"writes": [...], "seed": ...}` per channel. If no seed is
+   found (walk reaches root), `"seed"` is omitted — the consumer treats
+   this as "start empty."
+
+**Default implementation:**
+`BaseCheckpointSaver.aget_delta_channel_history` at
+`libs/checkpoint/langgraph/checkpoint/base/__init__.py` — one `aget_tuple`
+call per ancestor. Correct but O(chain_length) round-trips.
+
+---
+
+## Part II — Read & Write Surface
+
+These are the saver methods that interact with `DeltaChannel`. The base
+class provides correct defaults; overrides exist only as performance
+optimizations.
+
+### `aput` — accepting `_DeltaSnapshot` in `channel_values`
+
+Snapshot blobs reach `aput` as `_DeltaSnapshot(value)` instances inside
+`checkpoint["channel_values"]`. Round-tripping them through
+`JsonPlusSerializer` is sufficient (msgpack ext code `EXT_DELTA_SNAPSHOT`).
+No special-casing is required for correctness.
+
+**Optimization (Postgres):** hoist non-primitive values into a side blob
+table (`checkpoint_blobs`) and replace the JSON value with `True`.
+
+**References:**
+- `InMemorySaver.put` — simplest, at `libs/checkpoint/langgraph/checkpoint/memory/__init__.py`
+- `PostgresSaver.put` — sentinel + side-blob, at `libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py`
+- `SqliteSaver.put` — inline, at `libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py`
+
+### `aput_writes` — appending pending writes
+
+One row per `(channel, idx)` keyed by
+`(thread_id, checkpoint_ns, checkpoint_id, task_id, idx)`. No
+DeltaChannel-specific logic needed — these rows are what the parent-chain
+walk reads.
+
+### `aget_tuple` and `alist` — joining writes
+
+Must return `pending_writes` populated (oldest-first per task) so the
+default `aget_delta_channel_history` walk can collect them from each
+ancestor.
+
+### `aget_delta_channel_history` — performance optimization
+
+The base implementation works out of the box but does one `aget_tuple` per
+ancestor. For long chains (up to `snapshot_frequency` = 1000 steps), this
+becomes expensive. Override for performance.
+
+**Working references (progressively richer):**
+- **InMemory** — walks dict keys directly:
+  `libs/checkpoint/langgraph/checkpoint/memory/__init__.py`
+- **SQLite** — paged DESC scan + per-channel UNION ALL writes fetch:
+  `libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/_delta.py` and
+  `libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py`
+- **Postgres** — JSONB-aware two-stage SQL (stage 1: paged `checkpoint_id`
+  DESC with parallel `channel_values->key IS NOT NULL` checks; stage 2:
+  per-channel UNION ALL writes + seed blob fetch):
+  `libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py` and
+  `libs/checkpoint-postgres/langgraph/checkpoint/postgres/__init__.py`
+
+### `adelete_thread` — safe cleanup
+
+Wholesale delete of all rows for a `thread_id`. No DeltaChannel risk
+because nothing survives. Copy the pattern from any OSS saver.
+
+---
+
+## Part III — Cleanup Surface
+
+These optional methods (`prune`, `delete_for_runs`, `copy_thread`) have
+**no OSS implementation** yet. The docstrings on `BaseCheckpointSaver` flag
+the danger; this section provides concrete recipes.
+
+### The pitfall
+
+- The "latest" checkpoint is rarely a snapshot point (`snapshot_frequency`
+  defaults to 1000).
+- A naive `keep_latest` that drops intermediate ancestors and their
+  `checkpoint_writes` severs the parent chain.
+- The surviving head's reconstruction walk hits `parent_config = None`
+  early, finds no seed, and `from_checkpoint(MISSING)` produces an empty
+  value. **Silent data loss — no exception raised.**
+- The same trap applies to `delete_for_runs` (a run's writes may be
+  ancestors of a live head) and `copy_thread` (head-only copy strands the
+  target).
+
+### The `get_delta_channel_keepset` helper
+
+`BaseCheckpointSaver` provides a helper that returns the minimum set of
+`checkpoint_id`s that must survive deletion:
+
+```python
+keep = await saver.aget_delta_channel_keepset(
+    config=head_config, channels=["messages", "events"],
+)
+# `keep` contains the head + every ancestor back to the nearest
+# _DeltaSnapshot for each listed channel. Delete everything else.
+```
+
+Pass `channels=[]` for graphs without `DeltaChannel` — returns
+`{head_id}` only.
+
+### Implementing `prune`
+
+Three strategies, ordered by recommendation:
+
+#### Strategy A — Compact-and-prune (requires graph access)
+
+1. Resolve current value via
+   `saver.aget_delta_channel_history(config=head, channels=delta_channels)`.
+2. Fold writes through each channel's `reducer` to get the live value.
+3. Rewrite `channel_values[ch] = _DeltaSnapshot(value)` on the kept head
+   (update the checkpoint row + blob table).
+4. Now `aget_delta_channel_keepset(channels=[])` returns `{head_id}` —
+   delete everything else.
+
+This is the most space-efficient but requires access to the graph's
+reducers, so it must be done at the application layer (not purely
+saver-side).
+
+#### Strategy B — Walk-to-boundary (purely saver-side, recommended)
+
+```python
+async def aprune(self, thread_ids, *, strategy="keep_latest"):
+    for tid in thread_ids:
+        heads = await self._list_heads(tid)         # your backend code
+        keep: set[str] = set()
+        for h in heads:
+            keep |= await self.aget_delta_channel_keepset(
+                config=h, channels=DELTA_CHANNELS,
+            )
+        all_ids = await self._list_all_ids(tid)     # your backend code
+        to_delete = all_ids - keep
+        await self._delete_checkpoint_rows(tid, to_delete)
+        await self._delete_writes(tid, to_delete)
+```
+
+No rewrite needed. Preserves more rows than Strategy A but is correct and
+simple.
+
+**Postgres SQL sketch** (alternative to the Python helper — pure SQL
+recursive CTE):
+
+```sql
+WITH RECURSIVE ancestors AS (
+    SELECT checkpoint_id, parent_checkpoint_id,
+           (checkpoint -> 'channel_values' -> 'messages') IS NOT NULL AS has_snap
+    FROM checkpoints
+    WHERE thread_id = $1 AND checkpoint_ns = $2 AND checkpoint_id = $3
+
+    UNION ALL
+
+    SELECT c.checkpoint_id, c.parent_checkpoint_id,
+           (c.checkpoint -> 'channel_values' -> 'messages') IS NOT NULL
+    FROM checkpoints c
+    JOIN ancestors a ON c.checkpoint_id = a.parent_checkpoint_id
+    WHERE NOT a.has_snap
+)
+SELECT checkpoint_id FROM ancestors;
+-- This returns the keep-set for channel 'messages'.
+```
+
+#### Strategy C — Refuse
+
+Detect `DeltaChannel` usage (e.g. check
+`delta_updates_since_snapshot` in metadata) and skip pruning those threads.
+Cheapest correct option for projects that don't need pruning yet.
+
+### Implementing `copy_thread`
+
+**Default safe approach:** copy *all* rows for
+`(source_thread_id, *)` to `(target_thread_id, *)` across all three
+tables. Single transaction, simple `INSERT ... SELECT`. No
+DeltaChannel-specific logic needed — the full chain is preserved.
+
+```sql
+-- Postgres example
+INSERT INTO checkpoints (thread_id, checkpoint_ns, checkpoint_id, ...)
+SELECT $target, checkpoint_ns, checkpoint_id, ...
+FROM checkpoints WHERE thread_id = $source;
+
+INSERT INTO checkpoint_blobs (thread_id, checkpoint_ns, channel, version, ...)
+SELECT $target, checkpoint_ns, channel, version, ...
+FROM checkpoint_blobs WHERE thread_id = $source;
+
+INSERT INTO checkpoint_writes (thread_id, checkpoint_ns, checkpoint_id, ...)
+SELECT $target, checkpoint_ns, checkpoint_id, ...
+FROM checkpoint_writes WHERE thread_id = $source;
+```
+
+**Optimization:** if you must copy only a sub-range, apply Strategy A
+(compact a snapshot onto the chosen head first) then copy that single
+checkpoint.
+
+### Implementing `delete_for_runs`
+
+`run_id` lives in `metadata` JSON, not as a column. Implementations need a
+JSON predicate:
+
+```sql
+-- Postgres
+SELECT checkpoint_id FROM checkpoints
+WHERE thread_id = $1 AND metadata->>'run_id' = ANY($2);
+```
+
+**Risk:** deleting a run's checkpoints/writes can strand surviving heads in
+the same thread whose reconstruction walk would have traversed those rows.
+
+**Recipe:**
+1. Identify all `(thread_id, checkpoint_id)` belonging to the run.
+2. For every still-live head in those threads, compute the keep-set:
+   `keep |= await saver.aget_delta_channel_keepset(config=head, channels=...)`
+3. Only delete run rows that fall **outside** the union of all keep-sets.
+4. If a row is both "belongs to run" and "in keep-set," it must survive.
+
+---
+
+## Part IV — Validation
+
+### Conformance suite
+
+Run the existing conformance tests for any methods you implement:
+
+```python
+from langgraph.checkpoint.conformance import validate
+
+report = await validate(my_checkpointer, capabilities={
+    "delta_channel_history",
+    "delta_channel_keepset",
+    "delta_channel_reconstruction",
+})
+report.print_report()
+assert report.passed_all_base()
+```
+
+The three delta capabilities test:
+- **`delta_channel_history`** — walk contract (writes oldest→newest, seed
+  is nearest populated `channel_values[ch]`, target's writes excluded).
+- **`delta_channel_keepset`** — keep-set contract (empty channels →
+  `{target}`, snapshot N-back → full chain, multi-channel → union).
+- **`delta_channel_reconstruction`** — end-to-end round-trip: write
+  `_DeltaSnapshot` + writes via `aput` / `aput_writes`, read via
+  `aget_delta_channel_history`, reconstruct via
+  `DeltaChannel.from_checkpoint(seed) + replay_writes(writes)`, assert
+  value equality.
+
+### Recommended additional test
+
+Build a graph with `DeltaChannel(snapshot_frequency=10)`, drive it for
+~25 supersteps, run your `prune` / `delete_for_runs` / `copy_thread`, then
+assert reconstructed value equals the pre-cleanup value. This is the single
+test that catches the silent-corruption mode (conformance alone is
+necessary but not sufficient for cleanup methods).
+
+See existing patterns in:
+- `libs/langgraph/tests/test_delta_channel_migration.py`
+- `libs/langgraph/tests/test_delta_channel_exit_mode.py`
+
+---
+
+## Part V — Future Work
+
+- **`DeltaChannelAwarePruneMixin`** — wraps `aget_delta_channel_keepset`
+  together with abstract backend hooks into ready-to-use `aprune` /
+  `acopy_thread` / `adelete_for_runs` bodies. Deferred because the hook
+  shape is opinionated and the keepset helper already factors out the
+  dangerous logic.
+
+  Sketch:
+  ```python
+  class DeltaChannelAwarePruneMixin:
+      DELTA_CHANNELS: ClassVar[Sequence[str]] = ()
+
+      async def _alist_thread_heads(self, tid: str) -> list[RunnableConfig]: ...
+      async def _alist_thread_checkpoint_ids(self, tid: str) -> set[str]: ...
+      async def _adelete_checkpoint_rows(self, tid: str, ids: set[str]) -> None: ...
+
+      async def aprune(self, thread_ids, *, strategy="keep_latest"):
+          for tid in thread_ids:
+              heads = await self._alist_thread_heads(tid)
+              keep: set[str] = set()
+              for h in heads:
+                  keep |= await self.aget_delta_channel_keepset(
+                      config=h, channels=self.DELTA_CHANNELS,
+                  )
+              all_ids = await self._alist_thread_checkpoint_ids(tid)
+              await self._adelete_checkpoint_rows(tid, all_ids - keep)
+  ```
+
+- **OSS `prune` / `delete_for_runs` / `copy_thread` implementations** on
+  Postgres / SQLite savers. Planned as separate PRs.
+
+- **Pregel-side Strategy A compaction helper** — needs graph + reducers;
+  lives in `langgraph` package, not the saver.
+
+- **API stabilization** — all methods marked Beta here will move to stable
+  once the design proves out in production.
+
+---
+
+## Cross-references
+
+- Docstring warnings on `BaseCheckpointSaver`: `prune`, `aprune`,
+  `delete_for_runs`, `adelete_for_runs`, `copy_thread`, `acopy_thread`
+  (`libs/checkpoint/langgraph/checkpoint/base/__init__.py`).
+- `DeltaChannel` Beta warning
+  (`libs/langgraph/langgraph/channels/delta.py`).
+- Conformance suite
+  (`libs/checkpoint-conformance/langgraph/checkpoint/conformance/`).

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_delta_channel_keepset",
+    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,98 @@
+"""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.serde.types import _DeltaSnapshot
+
+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
+
+    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,183 @@
+"""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())
+    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, 4], f"Expected [1,2,3,4], 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())
+    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, 5], f"Expected [4,5], 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']}"
+
+
+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,
+]
+
+
+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,166 @@
+"""DELTA_CHANNEL_RECONSTRUCTION capability tests — end-to-end round-trip.
+
+Exercises: aput + aput_writes + aget_delta_channel_history + from_checkpoint +
+replay_writes. This catches the most common silent-corruption mode: failing to
+round-trip `_DeltaSnapshot` blobs through serialization.
+"""
+
+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
+
+
+def _list_reducer(state: list, writes: list) -> list:
+    """Simple append reducer for testing."""
+    return state + writes
+
+
+async def test_reconstruction_basic(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Reconstruct delta channel value from history matches expected."""
+    from langgraph.channels.delta import DeltaChannel
+
+    tid = str(uuid4())
+    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")
+    from langgraph._internal._typing import MISSING
+
+    if seed is None:
+        seed = MISSING
+
+    ch = DeltaChannel(_list_reducer, list)
+    replay_ch = ch.from_checkpoint(seed)
+    replay_ch.replay_writes(history["writes"])
+    reconstructed = replay_ch.get()
+
+    # Expected: snapshot at step 0 = [0], then writes at steps 1,2,3,4
+    expected = [0] + [1] + [2] + [3] + [4]
+    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."""
+    from langgraph.channels.delta import DeltaChannel
+
+    tid = str(uuid4())
+    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")
+    from langgraph._internal._typing import MISSING
+
+    if seed is None:
+        seed = MISSING
+
+    ch = DeltaChannel(_list_reducer, list)
+    replay_ch = ch.from_checkpoint(seed)
+    replay_ch.replay_writes(history["writes"])
+    reconstructed = replay_ch.get()
+
+    # Snapshot at step 3 = [3], writes at steps 4,5
+    expected = [3] + [4] + [5]
+    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."""
+    from langgraph.channels.delta import DeltaChannel
+
+    tid = str(uuid4())
+    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"]
+
+    from langgraph._internal._typing import MISSING
+
+    seed = history.get("seed", MISSING)
+
+    ch = DeltaChannel(_list_reducer, list)
+    replay_ch = ch.from_checkpoint(seed)
+    replay_ch.replay_writes(history["writes"])
+    reconstructed = replay_ch.get()
+
+    expected = [0] + [1] + [2] + [3]
+    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,9 +43,14 @@ 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"
+missing-typed-dict-key = "ignore"
 
 [tool.ruff]
 lint.select = [

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

@@ -680,6 +680,104 @@ 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)
+
+        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)
+        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)
+        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.