chore: remove keepset helper (#7745)

Comment follow up

libs/checkpoint-conformance/README.md

@@ -64,8 +64,6 @@ The suite tests **base** capabilities (required) and **extended** capabilities (
 | `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

@@ -24,8 +24,6 @@ class Capability(str, Enum):
     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.
@@ -46,8 +44,6 @@ EXTENDED_CAPABILITIES = frozenset(
         Capability.COPY_THREAD,
         Capability.PRUNE,
         Capability.DELTA_CHANNEL_HISTORY,
-        Capability.DELTA_CHANNEL_KEEPSET,
-        Capability.DELTA_CHANNEL_RECONSTRUCTION,
     }
 )
 
@@ -64,8 +60,6 @@ _CAPABILITY_METHOD_MAP: dict[Capability, str] = {
     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

@@ -12,12 +12,6 @@ from langgraph.checkpoint.conformance.spec.test_delete_thread import (
 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
@@ -34,6 +28,4 @@ __all__ = [
     "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/test_delta_channel_keepset.py

@@ -1,172 +0,0 @@
-"""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

@@ -1,164 +0,0 @@
-"""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

@@ -22,12 +22,6 @@ from langgraph.checkpoint.conformance.spec.test_delete_thread import (
 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
@@ -45,8 +39,6 @@ _RUNNERS = {
     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-sqlite/tests/test_conformance_delta.py

@@ -27,8 +27,6 @@ async def test_delta_channel_conformance():
         sqlite_saver,
         capabilities={
             "delta_channel_history",
-            "delta_channel_keepset",
-            "delta_channel_reconstruction",
         },
     )
     for cap, result in report.results.items():

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

@@ -680,123 +680,6 @@ 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

@@ -25,8 +25,6 @@ async def test_delta_channel_conformance():
         mem_saver,
         capabilities={
             "delta_channel_history",
-            "delta_channel_keepset",
-            "delta_channel_reconstruction",
         },
     )
     for cap, result in report.results.items():