diff --git a/libs/checkpoint-conformance/README.md b/libs/checkpoint-conformance/README.md index 61807e60b..cc26a2ede 100644 --- a/libs/checkpoint-conformance/README.md +++ b/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. diff --git a/libs/checkpoint-conformance/langgraph/checkpoint/conformance/capabilities.py b/libs/checkpoint-conformance/langgraph/checkpoint/conformance/capabilities.py index a7b506a01..b65f85de1 100644 --- a/libs/checkpoint-conformance/langgraph/checkpoint/conformance/capabilities.py +++ b/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", } diff --git a/libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/__init__.py b/libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/__init__.py index 915964b5d..edcf3d07b 100644 --- a/libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/__init__.py +++ b/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", ] diff --git a/libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/test_delta_channel_keepset.py b/libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/test_delta_channel_keepset.py deleted file mode 100644 index 2311f04d9..000000000 --- a/libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/test_delta_channel_keepset.py +++ /dev/null @@ -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 diff --git a/libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/test_delta_channel_reconstruction.py b/libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/test_delta_channel_reconstruction.py deleted file mode 100644 index 5cd2760bd..000000000 --- a/libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/test_delta_channel_reconstruction.py +++ /dev/null @@ -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 diff --git a/libs/checkpoint-conformance/langgraph/checkpoint/conformance/validate.py b/libs/checkpoint-conformance/langgraph/checkpoint/conformance/validate.py index e1019e6a6..2f2945449 100644 --- a/libs/checkpoint-conformance/langgraph/checkpoint/conformance/validate.py +++ b/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, } diff --git a/libs/checkpoint-sqlite/tests/test_conformance_delta.py b/libs/checkpoint-sqlite/tests/test_conformance_delta.py index 0d8e83212..ba0e90f18 100644 --- a/libs/checkpoint-sqlite/tests/test_conformance_delta.py +++ b/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(): diff --git a/libs/checkpoint/langgraph/checkpoint/base/__init__.py b/libs/checkpoint/langgraph/checkpoint/base/__init__.py index e588a6073..7b727bc1e 100644 --- a/libs/checkpoint/langgraph/checkpoint/base/__init__.py +++ b/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. diff --git a/libs/checkpoint/tests/test_conformance_delta.py b/libs/checkpoint/tests/test_conformance_delta.py index 5afe330a8..82a52a93e 100644 --- a/libs/checkpoint/tests/test_conformance_delta.py +++ b/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():