fix: order delta channel replay by task path

libs/checkpoint-conformance/langgraph/checkpoint/conformance/spec/test_delta_channel_history.py

@@ -1,3 +1,3 @@
 """DELTA_CHANNEL_HISTORY capability tests — aget_delta_channel_history contract."""
 
 from __future__ import annotations
@@ -208,6 +208,93 @@
     assert values == [2], f"Expected [2], got {values}"
 
 
+# Task ids used by the ordering tests below. `build_delta_chain` tags its own
+# writes with a `uuid4`, whose hex digits are all <= "f", so "aaaa..." sorts
+# before every fixture task id and "zzzz..." sorts after every one of them.
+# That makes the expected order fully determined rather than dependent on which
+# uuid4 the fixture happened to draw.
+TASK_ID_SORTS_FIRST = "aaaaaaaa-0000-0000-0000-000000000000"
+TASK_ID_SORTS_LAST = "zzzzzzzz-0000-0000-0000-000000000000"
+
+
+async def test_history_orders_parallel_writes_by_task_path(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Writes from several tasks in one super-step replay in task_path order.
+
+    Live execution sorts a super-step's tasks by `task_path_str(path[:3])`
+    before applying their values, so replay has to recover that order rather
+    than `task_id` order — `task_id` is a hash of the path, so the two
+    disagree, and reducers are only required to be batching-invariant, not
+    order-invariant.
+
+    The two task_ids are assigned so they sort in the *opposite* order from
+    their task_paths. A saver ordering by `(task_id, idx)` therefore returns
+    these writes reversed, rather than passing by happening to agree.
+    """
+    configs = await build_delta_chain(
+        saver,
+        thread_id=str(uuid4()),
+        channel="ch",
+        snapshots_at_steps=[0],
+        total_steps=3,
+    )
+    # The chain is: step 0 snapshot (seed), step 1 write, step 2 write.
+    # `aget_delta_channel_history` walks from the head's parent back to the
+    # seed, so it collects step 1's writes only — step 0 terminates the walk
+    # and step 2 is the head, whose own writes are pending for the next
+    # super-step and excluded. So step 1 is where these writes have to go.
+    step_1, head = configs[1], configs[2]
+    await saver.aput_writes(
+        step_1, [("ch", "second")], TASK_ID_SORTS_FIRST, "~pull, 02"
+    )
+    await saver.aput_writes(step_1, [("ch", "first")], TASK_ID_SORTS_LAST, "~pull, 01")
+
+    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
+    values = [w[2] for w in result["ch"]["writes"]]
+    # 1 is the fixture's own write at step 1. It carries no task_path, so it
+    # sorts ahead of both writes added above.
+    assert values == [1, "first", "second"], (
+        f"Expected task_path order [1, 'first', 'second'], got {values}. "
+        "Ordering by (task_id, idx) alone yields [1, 'second', 'first']."
+    )
+
+
+async def test_history_orders_pathless_writes_first(
+    saver: BaseCheckpointSaver,
+) -> None:
+    """Writes stored without a task_path sort ahead of path-carrying ones.
+
+    A task-less write (graph input) persists `task_path=""`, as does any row
+    written before a saver recorded the column. `""` precedes every
+    `task_path_str` output because that function prefixes tuples with `~`, so
+    those writes replay first — where live execution applies graph input.
+    """
+    configs = await build_delta_chain(
+        saver,
+        thread_id=str(uuid4()),
+        channel="ch",
+        snapshots_at_steps=[0],
+        total_steps=3,
+    )
+    # Same chain shape as above: step 1 is the only step the walk collects.
+    step_1, head = configs[1], configs[2]
+    # Committed in the opposite order to the one they must replay in, so the
+    # assertion cannot pass on insertion order alone.
+    await saver.aput_writes(
+        step_1, [("ch", "from_node")], TASK_ID_SORTS_FIRST, "~pull, a"
+    )
+    await saver.aput_writes(step_1, [("ch", "from_input")], TASK_ID_SORTS_LAST)
+
+    result = await saver.aget_delta_channel_history(config=head, channels=["ch"])
+    values = [w[2] for w in result["ch"]["writes"]]
+    # Both 1 (the fixture's write) and "from_input" are pathless, so they sort
+    # by task_id among themselves and both precede the path-carrying write.
+    assert values == [1, "from_input", "from_node"], (
+        f"Expected pathless writes first, got {values}"
+    )
+
+
 ALL_DELTA_CHANNEL_HISTORY_TESTS = [
     test_history_returns_writes_oldest_first,
     test_history_seed_is_nearest_snapshot,
@@ -216,6 +303,8 @@
     test_history_empty_channels_returns_empty,
     test_history_walk_to_root_no_seed,
     test_history_migration_plain_value_as_seed,
+    test_history_orders_parallel_writes_by_task_path,
+    test_history_orders_pathless_writes_first,
 ]
 
 

libs/checkpoint-postgres/langgraph/checkpoint/postgres/base.py

@@ -168,6 +168,7 @@ class _DeltaStage2Row(TypedDict, total=False):
     type: str | None
     blob: bytes | None
     task_id: str | None  # "w" rows only
+    task_path: str | None  # "w" rows only
     idx: int | None  # "w" rows only
     version: str | None  # "b" rows only
 
@@ -272,15 +273,16 @@ def _build_delta_stage2_sql(
         branches.append(
             "SELECT 'w'::text AS _kind, "
             "checkpoint_id, channel, "
-            "type, blob, task_id, idx, NULL::text AS version "
+            "type, blob, task_id, task_path, idx, NULL::text AS version "
             "FROM checkpoint_writes "
             "WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s "
             "AND checkpoint_id = ANY(%s)"
         )
     for _ in channels_with_seed:
         branches.append(
             "SELECT 'b'::text AS _kind, NULL::text AS checkpoint_id, channel, "
-            "type, blob, NULL::text AS task_id, NULL::int AS idx, version "
+            "type, blob, NULL::text AS task_id, NULL::text AS task_path, "
+            "NULL::int AS idx, version "
             "FROM checkpoint_blobs "
             "WHERE thread_id = %s AND checkpoint_ns = %s AND channel = %s "
             "AND version = %s"
@@ -426,10 +428,11 @@ def _build_delta_channels_writes_history(
         seed blob is sentinel "empty" — in both cases the consumer treats
         absence as "start empty".
         """
-        # writes_by_ch_by_cid[channel][cid] = list of (type, blob, task_id, idx)
-        writes_by_ch_by_cid: dict[str, dict[str, list[tuple[str, bytes, str, int]]]] = {
-            ch: {} for ch in channels
-        }
+        # writes_by_ch_by_cid[channel][cid] = list of
+        # (type, blob, task_id, idx, task_path)
+        writes_by_ch_by_cid: dict[
+            str, dict[str, list[tuple[str, bytes, str, int, str]]]
+        ] = {ch: {} for ch in channels}
         # seed_blob_by_ver[(channel, version)] = (type, blob)
         seed_blob_by_ver: dict[tuple[str, str], tuple[str, bytes]] = {}
 
@@ -440,8 +443,17 @@ def _build_delta_channels_writes_history(
                 cid = cast(str, r["checkpoint_id"])
                 writes_by_ch_by_cid.setdefault(ch, {}).setdefault(cid, []).append(
                     cast(
-                        "tuple[str, bytes, str, int]",
-                        (r["type"], r["blob"], r["task_id"], r["idx"]),
+                        "tuple[str, bytes, str, int, str]",
+                        (
+                            r["type"],
+                            r["blob"],
+                            r["task_id"],
+                            r["idx"],
+                            # `task_path` is NOT NULL DEFAULT '' on "w" rows;
+                            # it is nullable on `_DeltaStage2Row` only because
+                            # the seed branch selects NULL for it.
+                            r["task_path"],
+                        ),
                     )
                 )
             else:  # kind == "b"
@@ -450,10 +462,12 @@ def _build_delta_channels_writes_history(
                     "tuple[str, bytes]", (r["type"], r["blob"])
                 )
 
-        # Sort writes per (channel, cid) newest-first by (task_id, idx)
+        # Sort writes per (channel, cid) newest-first by
+        # (task_path, task_id, idx) — the order `apply_writes` applied them
+        # in live, and the order documented on `DeltaChannelHistory`.
         for cid_map in writes_by_ch_by_cid.values():
             for ws in cid_map.values():
-                ws.sort(key=lambda w: (w[2], w[3]), reverse=True)
+                ws.sort(key=lambda w: (w[4], w[2], w[3]), reverse=True)
 
         result: dict[str, DeltaChannelHistory] = {}
         for ch in channels:
@@ -463,7 +477,9 @@ def _build_delta_channels_writes_history(
             collected: list[PendingWrite] = []
             cid_writes = writes_by_ch_by_cid.get(ch, {})
             for cid in chain_cids:
-                for type_tag, write_blob, task_id, _idx in cid_writes.get(cid, []):
+                for type_tag, write_blob, task_id, _idx, _path in cid_writes.get(
+                    cid, []
+                ):
                     val = self.serde.loads_typed((type_tag, write_blob))
                     collected.append((task_id, ch, val))
             collected.reverse()

libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/__init__.py

@@ -29,6 +29,11 @@
     build_delta_stage2_sql,
     step_walk_with_row,
 )
+from langgraph.checkpoint.sqlite._schema import (
+    ADD_WRITES_TASK_PATH_SQL,
+    DUPLICATE_COLUMN_ERROR,
+    HAS_WRITES_TASK_PATH_SQL,
+)
 from langgraph.checkpoint.sqlite.utils import search_where
 
 _AIO_ERROR_MSG = (
@@ -154,6 +159,7 @@ def setup(self) -> None:
                 checkpoint_ns TEXT NOT NULL DEFAULT '',
                 checkpoint_id TEXT NOT NULL,
                 task_id TEXT NOT NULL,
+                task_path TEXT NOT NULL DEFAULT '',
                 idx INTEGER NOT NULL,
                 channel TEXT NOT NULL,
                 type TEXT,
@@ -162,6 +168,12 @@ def setup(self) -> None:
             );
             """
         )
+        if not self.conn.execute(HAS_WRITES_TASK_PATH_SQL).fetchone():
+            try:
+                self.conn.execute(ADD_WRITES_TASK_PATH_SQL)
+            except sqlite3.OperationalError as exc:
+                if DUPLICATE_COLUMN_ERROR not in str(exc):
+                    raise
 
         self.is_setup = True
 
@@ -460,9 +472,9 @@ def put_writes(
             task_path: Path of the task creating the writes.
         """
         query = (
-            "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
+            "INSERT OR REPLACE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
             if all(w[0] in WRITES_IDX_MAP for w in writes)
-            else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?)"
+            else "INSERT OR IGNORE INTO writes (thread_id, checkpoint_ns, checkpoint_id, task_id, task_path, idx, channel, type, value) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)"
         )
         with self.cursor() as cur:
             cur.executemany(
@@ -473,6 +485,7 @@ def put_writes(
                         str(config["configurable"]["checkpoint_ns"]),
                         str(config["configurable"]["checkpoint_id"]),
                         task_id,
+                        task_path,
                         WRITES_IDX_MAP.get(channel, idx),
                         channel,
                         *self.serde.dumps_typed(value),
@@ -568,7 +581,7 @@ def get_delta_channel_history(
                     )
                 cur.execute(stage2_sql, stage2_params)
                 stage2_rows = cast(
-                    "list[tuple[str, str, str, int, str, bytes]]", cur.fetchall()
+                    "list[tuple[str, str, str, int, str, bytes, str]]", cur.fetchall()
                 )
             else:
                 stage2_rows = []

libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/_delta.py

@@ -57,7 +57,7 @@ def build_delta_stage2_sql(*, chain_lens: Sequence[int]) -> str:
     for n in chain_lens:
         cid_placeholders = ",".join("?" * n)
         branches.append(
-            "SELECT checkpoint_id, channel, task_id, idx, type, value "
+            "SELECT checkpoint_id, channel, task_id, idx, type, value, task_path "
             "FROM writes "
             "WHERE thread_id = ? AND checkpoint_ns = ? AND channel = ? "
             f"AND checkpoint_id IN ({cid_placeholders})"
@@ -130,29 +130,33 @@ def build_delta_channels_writes_history(
     chain_by_ch: Mapping[str, list[str]],
     seed_val_by_ch: Mapping[str, Any],
     seeded: set[str],
-    stage2_rows: Sequence[tuple[str, str, str, int, str, bytes]],
+    stage2_rows: Sequence[tuple[str, str, str, int, str, bytes, str]],
     serde: Any,
 ) -> dict[str, DeltaChannelHistory]:
     """Demux stage-2 rows per channel; produce per-channel histories.
 
-    Stage-2 rows are `(checkpoint_id, channel, task_id, idx, type, value)`.
-    Final write order is oldest→newest globally and `(task_id, idx)` within
-    a checkpoint, matching the contract on `DeltaChannelHistory.writes`.
+    Stage-2 rows are
+    `(checkpoint_id, channel, task_id, idx, type, value, task_path)`.
+    Final write order is oldest→newest globally and
+    `(task_path, task_id, idx)` within a checkpoint, matching the contract
+    on `DeltaChannelHistory.writes` — that is the order `apply_writes`
+    applied them in live, which `(task_id, idx)` alone does not recover
+    for parallel tasks writing one channel in a single super-step.
 
     `seed` is omitted when the walk reached a true root with no snapshot
     found (channel never entered `seeded`); consumers treat absence as
     "start empty".
     """
-    writes_by_ch_by_cid: dict[str, dict[str, list[tuple[str, bytes, str, int]]]] = {
-        ch: {} for ch in channels
-    }
-    for cid, ch, task_id, idx, type_tag, value_blob in stage2_rows:
+    writes_by_ch_by_cid: dict[
+        str, dict[str, list[tuple[str, bytes, str, int, str]]]
+    ] = {ch: {} for ch in channels}
+    for cid, ch, task_id, idx, type_tag, value_blob, task_path in stage2_rows:
         writes_by_ch_by_cid.setdefault(ch, {}).setdefault(cid, []).append(
-            (type_tag, value_blob, task_id, idx)
+            (type_tag, value_blob, task_id, idx, task_path)
         )
     for cid_map in writes_by_ch_by_cid.values():
         for ws in cid_map.values():
-            ws.sort(key=lambda w: (w[2], w[3]))
+            ws.sort(key=lambda w: (w[4], w[2], w[3]))
 
     result: dict[str, DeltaChannelHistory] = {}
     for ch in channels:
@@ -161,7 +165,7 @@ def build_delta_channels_writes_history(
         collected: list[PendingWrite] = []
         # Chain is newest-first; iterate oldest-first for the public order.
         for cid in reversed(chain_cids):
-            for type_tag, value_blob, task_id, _idx in cid_writes.get(cid, []):
+            for type_tag, value_blob, task_id, _idx, _path in cid_writes.get(cid, []):
                 collected.append(
                     (task_id, ch, serde.loads_typed((type_tag, value_blob)))
                 )

libs/checkpoint-sqlite/langgraph/checkpoint/sqlite/_schema.py

@@ -0,0 +1,37 @@
+"""Additive schema migrations shared by the sqlite savers.
+
+`SqliteSaver.setup` and `AsyncSqliteSaver.setup` create their tables with
+`CREATE TABLE IF NOT EXISTS`, which leaves a database created by an earlier
+version on the earlier schema. Sqlite has no `ADD COLUMN IF NOT EXISTS`
+(the postgres savers rely on that form), and re-running a plain
+`ALTER TABLE ... ADD COLUMN` raises `OperationalError: duplicate column
+name`. So each migration pairs an `ALTER` with a probe against
+`pragma_table_info` that tells us whether this database still needs it.
+
+Databases created fresh already carry every column from the `CREATE TABLE`
+statements, so the probe finds the column and the `ALTER` never runs.
+"""
+
+from __future__ import annotations
+
+# `writes.task_path` records the path of the task that produced a write.
+# Delta channel replay orders a checkpoint's writes by
+# (task_path, task_id, idx) to reproduce the order `apply_writes` applied
+# them in live; without the column, replay can only order by
+# (task_id, idx), which permutes writes made by parallel tasks in the same
+# super-step. Rows written before this migration keep the `''` default and
+# so sort ahead of path-carrying rows within their checkpoint.
+HAS_WRITES_TASK_PATH_SQL = (
+    "SELECT 1 FROM pragma_table_info('writes') WHERE name = 'task_path'"
+)
+
+ADD_WRITES_TASK_PATH_SQL = (
+    "ALTER TABLE writes ADD COLUMN task_path TEXT NOT NULL DEFAULT ''"
+)
+
+# Substring of the `OperationalError` sqlite raises when the column is already
+# there. The probe above is not enough on its own: two connections opening the
+# same file can both pass it and both issue the `ALTER`, and unlike
+# `CREATE TABLE IF NOT EXISTS` the loser of that race raises. Callers treat it
+# as success — whoever won did the same migration.
+DUPLICATE_COLUMN_ERROR = "duplicate column name"