mirror of
https://github.com/langchain-ai/langgraph.git
synced 2026-09-25 19:15:11 +02:00
fix: order delta channel replay by task path
DeltaChannel reconstructs its value by replaying ancestor writes through the reducer. Every saver ordered a checkpoint's writes by (task_id, idx), but live execution applies them in task-path order: apply_writes sorts a super-step's tasks by task_path_str(task.path[:3]) before calling channel.update. task_id is a hash of the path, so the two orders are unrelated, and two or more tasks writing one DeltaChannel in a single super-step replayed in an arbitrary permutation. Reducers are only required to be batching-invariant, not order-invariant, so the permutation changes the value: get_state disagreed with what invoke returned, and continuing the thread persisted the reordered replay as the base for later writes. Replay now orders by (task_path, task_id, idx), following the precedent already set for the Send channel by SELECT_PENDING_SENDS_SQL. InMemorySaver and the postgres savers already persisted task_path and only needed the sort key; sqlite accepted task_path on put_writes and dropped it, so the writes table gains the column, added by setup() to databases created by earlier versions. Writes stored without a task_path sort first within their checkpoint, which is where live execution applies the task-less input writes that carry "". Co-authored-by: ErenAta16 <149434812+ErenAta16@users.noreply.github.com> Co-authored-by: ragnarok268 <58264829+ragnarok268@users.noreply.github.com>
This commit is contained in:
co-authored by
ErenAta16
ragnarok268
parent
bdb85b5aa8
commit
cdd02084e9
@@ -0,0 +1,87 @@
|
||||
import sqlite3
|
||||
from pathlib import Path
|
||||
|
||||
import aiosqlite
|
||||
import pytest
|
||||
from langgraph.checkpoint.base import empty_checkpoint
|
||||
|
||||
from langgraph.checkpoint.sqlite import SqliteSaver
|
||||
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
|
||||
|
||||
WRITES_BEFORE_TASK_PATH = """
|
||||
CREATE TABLE writes (
|
||||
thread_id TEXT NOT NULL,
|
||||
checkpoint_ns TEXT NOT NULL DEFAULT '',
|
||||
checkpoint_id TEXT NOT NULL,
|
||||
task_id TEXT NOT NULL,
|
||||
idx INTEGER NOT NULL,
|
||||
channel TEXT NOT NULL,
|
||||
type TEXT,
|
||||
value BLOB,
|
||||
PRIMARY KEY (thread_id, checkpoint_ns, checkpoint_id, task_id, idx)
|
||||
);
|
||||
INSERT INTO writes VALUES ('t', '', 'c', 'old-task', 0, 'ch', 'null', X'');
|
||||
"""
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def legacy_db(tmp_path: Path) -> Path:
|
||||
db = tmp_path / "legacy.sqlite"
|
||||
with sqlite3.connect(db) as conn:
|
||||
conn.executescript(WRITES_BEFORE_TASK_PATH)
|
||||
return db
|
||||
|
||||
|
||||
def test_setup_migrates_legacy_writes_table_repeatably(legacy_db: Path) -> None:
|
||||
for _ in range(2):
|
||||
with SqliteSaver.from_conn_string(str(legacy_db)) as saver:
|
||||
saver.setup()
|
||||
rows = saver.conn.execute(
|
||||
"SELECT task_id, task_path FROM writes"
|
||||
).fetchall()
|
||||
assert rows == [("old-task", "")]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("fresh", [True, False], ids=["fresh", "legacy"])
|
||||
def test_put_writes_persists_task_path(
|
||||
tmp_path: Path, legacy_db: Path, fresh: bool
|
||||
) -> None:
|
||||
db = tmp_path / "fresh.sqlite" if fresh else legacy_db
|
||||
with SqliteSaver.from_conn_string(str(db)) as saver:
|
||||
config = saver.put(
|
||||
{"configurable": {"thread_id": "t", "checkpoint_ns": ""}},
|
||||
empty_checkpoint(),
|
||||
{},
|
||||
{},
|
||||
)
|
||||
saver.put_writes(config, [("ch", "v")], "task-1", "~__pregel_pull, node")
|
||||
stored = saver.conn.execute(
|
||||
"SELECT task_path FROM writes WHERE task_id = 'task-1'"
|
||||
).fetchall()
|
||||
assert stored == [("~__pregel_pull, node",)]
|
||||
|
||||
|
||||
async def test_async_setup_migrates_legacy_writes_table_repeatably(
|
||||
legacy_db: Path,
|
||||
) -> None:
|
||||
for _ in range(2):
|
||||
async with AsyncSqliteSaver.from_conn_string(str(legacy_db)) as saver:
|
||||
await saver.setup()
|
||||
config = await saver.aput(
|
||||
{"configurable": {"thread_id": "t", "checkpoint_ns": ""}},
|
||||
empty_checkpoint(),
|
||||
{},
|
||||
{},
|
||||
)
|
||||
await saver.aput_writes(
|
||||
config, [("ch", "v")], "task-1", "~__pregel_pull, node"
|
||||
)
|
||||
|
||||
async with aiosqlite.connect(legacy_db) as conn:
|
||||
async with conn.execute(
|
||||
"SELECT DISTINCT task_id, task_path FROM writes ORDER BY task_id"
|
||||
) as cur:
|
||||
assert await cur.fetchall() == [
|
||||
("old-task", ""),
|
||||
("task-1", "~__pregel_pull, node"),
|
||||
]
|
||||
Reference in New Issue
Block a user