mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-08-24 08:32:24 +02:00
398d6cc59d
## Summary Add a user-facing design doc and `get_delta_channel_keepset` helper for third-party `BaseCheckpointSaver` authors who need to support graphs using `DeltaChannel`. **Deliverables:** 1. ~~**`docs/delta-channel-checkpointer-guide.md`** — comprehensive guide covering~~: moved to docs repo 2. **`BaseCheckpointSaver.get_delta_channel_keepset` / `aget_delta_channel_keepset`** — returns the minimum set of ancestor `checkpoint_id`s that must survive deletion for a given head's `DeltaChannel` reconstruction to remain intact. Enables safe `prune` implementations without silently corrupting delta history. 3. **Docstring warnings** on `prune`, `aprune`, `delete_for_runs`, `adelete_for_runs`, `copy_thread`, `acopy_thread` explaining the DeltaChannel pitfall (silent data loss if ancestor writes/snapshots are deleted). 4. **Three new conformance capabilities** in `libs/checkpoint-conformance`: - `delta_channel_history` — validates the `aget_delta_channel_history` walk contract - `delta_channel_keepset` — validates the keep-set contract - `delta_channel_reconstruction` — end-to-end round-trip (aput + aput_writes + history + reconstruct) ## Test plan - [x] `make format lint` passes in `libs/checkpoint`, `libs/checkpoint-conformance` - [x] All three new conformance capabilities pass against `InMemorySaver` - [x] Run conformance against SQLite saver - [x] Run conformance against Postgres saver --------- Co-authored-by: Cursor <cursoragent@cursor.com>
38 lines
1.1 KiB
Python
38 lines
1.1 KiB
Python
"""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}")
|