add delta overwrite graph coverage

examples/delta-channel-dump/README.md

@@ -1,132 +0,0 @@
-# delta-channel-dump
-
-Recover messages (and other channels) from a Postgres-backed LangGraph thread
-written by **langgraph >= 1.2** (DeltaChannel format) — including **LangGraph
-Server / langgraph-api** deployments on the Postgres runtime, **deepagents
-0.6.x**, or any OSS app using `PostgresSaver` — before rolling back to an older
-runtime such as **deepagents 0.5.x / langgraph < 1.2**.
-
-langgraph-api uses the same `checkpoints` / `checkpoint_blobs` / `checkpoint_writes`
-schema as OSS `checkpoint-postgres`; this script reads those tables directly.
-
-On the older runtime, `add_messages` does not understand the `EXT_DELTA_SNAPSHOT`
-msgpack ext code and silently returns an empty list for affected channels. This
-tool reads the raw checkpoint blobs from Postgres and emits JSON you can inspect
-and re-apply via `update_state` (LangGraph Server SDK) or `graph.update_state`
-(OSS).
-
-## Install
-
-```bash
-pip install "psycopg[binary]" ormsgpack
-```
-
-## Run
-
-```bash
-export DATABASE_URI=postgres://user:pass@host:5432/dbname
-python3 dump.py \
-    --thread-id <uuid> \
-    --channel messages \
-    [--channel files ...] \
-    [--checkpoint-id <uuid>] \
-    [--checkpoint-ns ""] \
-    --output recovery.json
-```
-
-- `--thread-id` (required): thread UUID
-- `--channel` (required, repeatable): channel names to recover
-- `--checkpoint-id` (optional): target checkpoint; defaults to latest
-- `--checkpoint-ns` (optional): namespace; defaults to `""`
-- `--database-uri` (optional): Postgres URI; defaults to `DATABASE_URI` env var
-- `--output` (optional): output file; defaults to stdout
-
-## Output
-
-```json
-{
-  "thread_id": "...",
-  "checkpoint_ns": "",
-  "target_checkpoint_id": "...",
-  "parent_checkpoint_id": "...",
-  "channels": {
-    "messages": {
-      "delta_kind": "snapshot",
-      "seed_checkpoint_id": "...",
-      "seed_version": "...",
-      "seed": [{ "type": "ai", "content": "...", "id": "ai-0" }],
-      "writes": [
-        {
-          "checkpoint_id": "...",
-          "task_id": "...",
-          "idx": 0,
-          "value": [{ "type": "ai", "content": "...", "id": "ai-10" }]
-        }
-      ]
-    }
-  }
-}
-```
-
-`delta_kind` is one of:
-
-- `snapshot` — DeltaChannel snapshot blob (`channel_values[ch] == true`)
-- `legacy_plain` — pre-DeltaChannel inline or blob value
-- `no_seed` — walked to root without finding a populated ancestor
-
-`writes` are ordered oldest-to-newest (the order a reducer would replay them).
-
-## Reducing back to a single list
-
-For deepagents-style messages, combine seed and writes, then deduplicate:
-
-```python
-import json
-
-data = json.load(open("recovery.json"))
-ch = data["channels"]["messages"]
-messages = list(ch["seed"] or [])
-for w in ch["writes"]:
-    messages.extend(w["value"] or [])
-
-# Dedup by id, keep last; drop RemoveMessage tombstones
-by_id = {}
-for m in messages:
-    if isinstance(m, dict) and m.get("type") == "remove":
-        by_id.pop(m.get("id"), None)
-    elif isinstance(m, dict) and m.get("id"):
-        by_id[m["id"]] = m
-    else:
-        by_id[id(m)] = m
-reduced = list(by_id.values())
-```
-
-This approximates `_messages_delta_reducer` semantics; adjust for your graph.
-
-## Re-applying
-
-```python
-from langgraph_sdk import get_client
-
-client = get_client(url="http://localhost:8123")
-await client.threads.update_state(
-    thread_id,
-    values={"messages": reduced},
-)
-```
-
-Review the recovered JSON before calling `update_state`. This tool is
-read-only and intentionally does not mutate the database.
-
-## Scope / non-goals (v1)
-
-- **Postgres only** — OSS `PostgresSaver` or langgraph-api Postgres runtime; not
-  inmem, gRPC core, Mongo, or Redis checkpointer backends
-- **No AES decryption** (`LANGGRAPH_AES_KEY`) or custom encryption
-- **No reducer** — raw seed + writes only
-- **No automatic `update_state`** — operator applies manually
-
-## Copying
-
-`dump.py` is self-contained. Copy it anywhere; only `psycopg[binary]` and
-`ormsgpack` are required at runtime. No langgraph imports.

examples/delta-channel-dump/dump.py

@@ -1,469 +0,0 @@
-#!/usr/bin/env python3
-"""Recover delta-channel state from a Postgres-backed LangGraph thread.
-
-Works with OSS ``PostgresSaver`` and LangGraph Server / langgraph-api on the
-Postgres runtime (same checkpoint schema). Use after rolling back from
-langgraph >= 1.2 / deepagents 0.6.x to an older runtime that does not
-understand ``EXT_DELTA_SNAPSHOT`` msgpack blobs. The script walks the checkpoint
-parent chain, decodes msgpack blobs, and emits a JSON dump of per-channel
-``seed`` plus oldest-to-newest ``writes``. Apply the recovered values manually
-via ``client.threads.update_state(...)`` (Server) or ``graph.update_state``
-(OSS).
-
-Install::
-
-    pip install "psycopg[binary]" ormsgpack
-
-Run::
-
-    export DATABASE_URI=postgres://...
-    python3 dump.py --thread-id <uuid> --channel messages --output recovery.json
-
-Scope (v1): Postgres only; no AES/custom encryption; no reducer application.
-"""
-
-from __future__ import annotations
-
-import argparse
-import base64
-import json
-import os
-import sys
-import uuid
-from typing import Any
-
-import ormsgpack
-import psycopg
-
-# LangGraph msgpack EXT type codes (langgraph/checkpoint/serde/jsonplus.py).
-EXT_CONSTRUCTOR_SINGLE_ARG = 0
-EXT_CONSTRUCTOR_POS_ARGS = 1
-EXT_CONSTRUCTOR_KW_ARGS = 2
-EXT_METHOD_SINGLE_ARG = 3
-EXT_PYDANTIC_V1 = 4
-EXT_PYDANTIC_V2 = 5
-EXT_NUMPY_ARRAY = 6
-EXT_DELTA_SNAPSHOT = 7
-
-_MSGPACK_OPTION = ormsgpack.OPT_NON_STR_KEYS
-
-
-def ext_hook(code: int, data: bytes) -> Any:
-    """Decode LangGraph msgpack EXT payloads to JSON-friendly Python values."""
-    if code == EXT_DELTA_SNAPSHOT:
-        inner = ormsgpack.unpackb(data, ext_hook=ext_hook, option=_MSGPACK_OPTION)
-        return {"__delta_snapshot__": inner}
-    if code == EXT_CONSTRUCTOR_SINGLE_ARG:
-        try:
-            tup = ormsgpack.unpackb(data, ext_hook=ext_hook, option=_MSGPACK_OPTION)
-            if tup[0] == "uuid" and tup[1] == "UUID":
-                hex_ = tup[2]
-                return (
-                    f"{hex_[:8]}-{hex_[8:12]}-{hex_[12:16]}-"
-                    f"{hex_[16:20]}-{hex_[20:]}"
-                )
-            return tup[2]
-        except Exception:
-            return None
-    if code == EXT_CONSTRUCTOR_POS_ARGS:
-        try:
-            tup = ormsgpack.unpackb(data, ext_hook=ext_hook, option=_MSGPACK_OPTION)
-            if tup[0] == "langgraph.types" and tup[1] == "Send":
-                args = tup[2]
-                if len(args) == 2:
-                    return {"__send__": {"node": args[0], "arg": args[1]}}
-                return {
-                    "__send__": {
-                        "node": args[0],
-                        "arg": args[1],
-                        "timeout": args[2],
-                    }
-                }
-            return tup[2]
-        except Exception:
-            return None
-    if code in (EXT_CONSTRUCTOR_KW_ARGS, EXT_METHOD_SINGLE_ARG):
-        try:
-            tup = ormsgpack.unpackb(data, ext_hook=ext_hook, option=_MSGPACK_OPTION)
-            return tup[2]
-        except Exception:
-            return None
-    if code in (EXT_PYDANTIC_V1, EXT_PYDANTIC_V2):
-        try:
-            tup = ormsgpack.unpackb(data, ext_hook=ext_hook, option=_MSGPACK_OPTION)
-            return tup[2]
-        except Exception:
-            return None
-    if code == EXT_NUMPY_ARRAY:
-        try:
-            dtype_str, shape, order, buf = ormsgpack.unpackb(
-                data, ext_hook=ext_hook, option=_MSGPACK_OPTION
-            )
-            return {
-                "__numpy_array__": {
-                    "dtype": dtype_str,
-                    "shape": shape,
-                    "order": order,
-                    "data_b64": base64.b64encode(buf).decode("ascii"),
-                }
-            }
-        except Exception:
-            return None
-    return None
-
-
-def decode_blob(blob_type: str, blob_bytes: bytes | None) -> Any:
-    if blob_type in ("empty", "null") or blob_bytes is None:
-        return None
-    if blob_type == "msgpack":
-        return ormsgpack.unpackb(
-            blob_bytes, ext_hook=ext_hook, option=_MSGPACK_OPTION
-        )
-    if blob_type in ("bytes", "bytearray"):
-        return base64.b64encode(blob_bytes).decode("ascii")
-    raise RuntimeError(
-        f"Unknown blob type {blob_type!r}. "
-        "AES/custom-encrypted deployments are out of scope for v1."
-    )
-
-
-def delta_unwrap(value: Any) -> Any:
-    if isinstance(value, dict) and "__delta_snapshot__" in value:
-        return value["__delta_snapshot__"]
-    return value
-
-
-def json_default(obj: Any) -> Any:
-    if isinstance(obj, (bytes, bytearray)):
-        return base64.b64encode(bytes(obj)).decode("ascii")
-    if isinstance(obj, uuid.UUID):
-        return str(obj)
-    raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable")
-
-
-def resolve_target_checkpoint_id(
-    conn: psycopg.Connection[Any],
-    thread_id: str,
-    checkpoint_ns: str,
-    checkpoint_id: str | None,
-) -> str:
-    if checkpoint_id is not None:
-        return checkpoint_id
-    row = conn.execute(
-        """
-        SELECT checkpoint_id::text
-        FROM checkpoints
-        WHERE thread_id = %s AND checkpoint_ns = %s
-        ORDER BY checkpoint_id DESC
-        LIMIT 1
-        """,
-        (thread_id, checkpoint_ns),
-    ).fetchone()
-    if row is None:
-        raise SystemExit(
-            f"No checkpoints found for thread_id={thread_id!r} "
-            f"checkpoint_ns={checkpoint_ns!r}"
-        )
-    return row[0]
-
-
-def _load_checkpoint(
-    conn: psycopg.Connection[Any],
-    thread_id: str,
-    checkpoint_ns: str,
-    checkpoint_id: str,
-) -> tuple[dict[str, Any], str | None] | None:
-    row = conn.execute(
-        """
-        SELECT checkpoint, parent_checkpoint_id::text
-        FROM checkpoints
-        WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id = %s
-        """,
-        (thread_id, checkpoint_ns, checkpoint_id),
-    ).fetchone()
-    if row is None:
-        return None
-    return row[0], row[1]
-
-
-def _load_seed(
-    conn: psycopg.Connection[Any],
-    *,
-    thread_id: str,
-    checkpoint_ns: str,
-    channel: str,
-    checkpoint_id: str,
-    channel_values: dict[str, Any],
-    channel_versions: dict[str, str],
-) -> dict[str, Any]:
-    cv = channel_values[channel]
-    version = channel_versions.get(channel)
-    if cv is True:
-        blob_row = conn.execute(
-            """
-            SELECT type, blob
-            FROM checkpoint_blobs
-            WHERE thread_id = %s AND checkpoint_ns = %s
-              AND channel = %s AND version = %s
-            """,
-            (thread_id, checkpoint_ns, channel, version),
-        ).fetchone()
-        seed_value = None
-        if blob_row is not None and blob_row[0] != "empty":
-            seed_value = delta_unwrap(decode_blob(blob_row[0], blob_row[1]))
-        return {
-            "delta_kind": "snapshot",
-            "seed_checkpoint_id": checkpoint_id,
-            "seed_version": version,
-            "seed": seed_value,
-        }
-    if isinstance(cv, (int, float, str, bool)) or cv is None:
-        return {
-            "delta_kind": "legacy_plain",
-            "seed_checkpoint_id": checkpoint_id,
-            "seed_version": version,
-            "seed": cv,
-        }
-    blob_row = conn.execute(
-        """
-        SELECT type, blob
-        FROM checkpoint_blobs
-        WHERE thread_id = %s AND checkpoint_ns = %s
-          AND channel = %s AND version = %s
-        """,
-        (thread_id, checkpoint_ns, channel, version),
-    ).fetchone()
-    seed_value = cv if blob_row is None else decode_blob(blob_row[0], blob_row[1])
-    return {
-        "delta_kind": "legacy_plain",
-        "seed_checkpoint_id": checkpoint_id,
-        "seed_version": version,
-        "seed": seed_value,
-    }
-
-
-def _load_writes_for_checkpoint(
-    conn: psycopg.Connection[Any],
-    *,
-    thread_id: str,
-    checkpoint_ns: str,
-    checkpoint_id: str,
-    channel: str,
-) -> list[dict[str, Any]]:
-    """Load writes for one checkpoint, newest-first by ``(task_id, idx)``.
-
-    ``walk_channel`` reverses the accumulated flat list before returning;
-    DESC here yields oldest-first within each checkpoint in the final output,
-    matching ``PostgresSaver._build_delta_channels_writes_history``.
-    """
-    rows = conn.execute(
-        """
-        SELECT task_id::text, idx, type, blob
-        FROM checkpoint_writes
-        WHERE thread_id = %s AND checkpoint_ns = %s
-          AND checkpoint_id = %s AND channel = %s
-        ORDER BY task_id DESC, idx DESC
-        """,
-        (thread_id, checkpoint_ns, checkpoint_id, channel),
-    ).fetchall()
-    return [
-        {
-            "checkpoint_id": checkpoint_id,
-            "task_id": task_id,
-            "idx": idx,
-            "value": decode_blob(blob_type, blob_bytes),
-        }
-        for task_id, idx, blob_type, blob_bytes in rows
-    ]
-
-
-def walk_channel(
-    conn: psycopg.Connection[Any],
-    *,
-    thread_id: str,
-    checkpoint_ns: str,
-    start_checkpoint_id: str | None,
-    channel: str,
-) -> dict[str, Any]:
-    """Walk one channel's parent chain from target.parent backward to seed."""
-    chain_writes_newest_first: list[dict[str, Any]] = []
-    cur = start_checkpoint_id
-    while cur is not None:
-        loaded = _load_checkpoint(conn, thread_id, checkpoint_ns, cur)
-        if loaded is None:
-            break
-        checkpoint_json, parent_id = loaded
-        channel_values = checkpoint_json.get("channel_values") or {}
-        channel_versions = checkpoint_json.get("channel_versions") or {}
-
-        chain_writes_newest_first.extend(
-            _load_writes_for_checkpoint(
-                conn,
-                thread_id=thread_id,
-                checkpoint_ns=checkpoint_ns,
-                checkpoint_id=cur,
-                channel=channel,
-            )
-        )
-
-        if channel in channel_values:
-            seed = _load_seed(
-                conn,
-                thread_id=thread_id,
-                checkpoint_ns=checkpoint_ns,
-                channel=channel,
-                checkpoint_id=cur,
-                channel_values=channel_values,
-                channel_versions=channel_versions,
-            )
-            seed["writes"] = list(reversed(chain_writes_newest_first))
-            return seed
-
-        cur = parent_id
-
-    return {
-        "delta_kind": "no_seed",
-        "seed_checkpoint_id": None,
-        "seed_version": None,
-        "seed": None,
-        "writes": list(reversed(chain_writes_newest_first)),
-    }
-
-
-def walk_parent_chain(
-    conn: psycopg.Connection[Any],
-    *,
-    thread_id: str,
-    checkpoint_ns: str,
-    target_checkpoint_id: str,
-    channels: list[str],
-) -> tuple[str | None, dict[str, dict[str, Any]]]:
-    """Walk parent chain and return per-channel seed + writes (oldest-first)."""
-    parent_row = conn.execute(
-        """
-        SELECT parent_checkpoint_id::text
-        FROM checkpoints
-        WHERE thread_id = %s AND checkpoint_ns = %s AND checkpoint_id = %s
-        """,
-        (thread_id, checkpoint_ns, target_checkpoint_id),
-    ).fetchone()
-    if parent_row is None:
-        raise SystemExit(
-            f"Checkpoint {target_checkpoint_id!r} not found for thread {thread_id!r}"
-        )
-    parent_checkpoint_id = parent_row[0]
-
-    result = {
-        ch: walk_channel(
-            conn,
-            thread_id=thread_id,
-            checkpoint_ns=checkpoint_ns,
-            start_checkpoint_id=parent_checkpoint_id,
-            channel=ch,
-        )
-        for ch in channels
-    }
-    return parent_checkpoint_id, result
-
-
-def build_output(
-    *,
-    thread_id: str,
-    checkpoint_ns: str,
-    target_checkpoint_id: str,
-    parent_checkpoint_id: str | None,
-    channels: dict[str, dict[str, Any]],
-) -> dict[str, Any]:
-    return {
-        "thread_id": thread_id,
-        "checkpoint_ns": checkpoint_ns,
-        "target_checkpoint_id": target_checkpoint_id,
-        "parent_checkpoint_id": parent_checkpoint_id,
-        "channels": channels,
-    }
-
-
-def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
-    parser = argparse.ArgumentParser(
-        description=(
-            "Recover delta-channel seed + writes from Postgres checkpoint data."
-        ),
-    )
-    parser.add_argument("--thread-id", required=True, help="Thread UUID")
-    parser.add_argument(
-        "--channel",
-        action="append",
-        required=True,
-        dest="channels",
-        help="Channel name (repeatable)",
-    )
-    parser.add_argument(
-        "--checkpoint-id",
-        default=None,
-        help="Target checkpoint UUID (default: latest for thread)",
-    )
-    parser.add_argument(
-        "--checkpoint-ns",
-        default="",
-        help='Checkpoint namespace (default: "")',
-    )
-    parser.add_argument(
-        "--database-uri",
-        default=os.environ.get("DATABASE_URI"),
-        help="Postgres URI (default: DATABASE_URI env var)",
-    )
-    parser.add_argument(
-        "--output",
-        default="-",
-        help="Output JSON file path (default: stdout)",
-    )
-    return parser.parse_args(argv)
-
-
-def main(argv: list[str] | None = None) -> int:
-    args = parse_args(argv)
-    if not args.database_uri:
-        print(
-            "error: --database-uri or DATABASE_URI is required",
-            file=sys.stderr,
-        )
-        return 2
-
-    thread_id = str(uuid.UUID(args.thread_id))
-    channels = list(dict.fromkeys(args.channels))
-
-    with psycopg.connect(args.database_uri) as conn:
-        target_checkpoint_id = resolve_target_checkpoint_id(
-            conn,
-            thread_id,
-            args.checkpoint_ns,
-            args.checkpoint_id,
-        )
-        parent_checkpoint_id, channel_data = walk_parent_chain(
-            conn,
-            thread_id=thread_id,
-            checkpoint_ns=args.checkpoint_ns,
-            target_checkpoint_id=target_checkpoint_id,
-            channels=channels,
-        )
-
-    output = build_output(
-        thread_id=thread_id,
-        checkpoint_ns=args.checkpoint_ns,
-        target_checkpoint_id=target_checkpoint_id,
-        parent_checkpoint_id=parent_checkpoint_id,
-        channels=channel_data,
-    )
-    payload = json.dumps(output, indent=2, default=json_default)
-
-    if args.output == "-":
-        print(payload)
-    else:
-        with open(args.output, "w", encoding="utf-8") as f:
-            f.write(payload)
-            f.write("\n")
-    return 0
-
-
-if __name__ == "__main__":
-    raise SystemExit(main())

libs/langgraph/langgraph/channels/delta.py

@@ -177,8 +177,7 @@ class DeltaChannel(Generic[Value], BaseChannel[Any, Any, Any]):
                 if overwrite_value is not None
                 else self.typ()
             )
-            remaining = [v for i, v in enumerate(values) if i != overwrite_idx]
-            self.value = self.reducer(base, remaining) if remaining else base
+            self.value = base
             return True
         base = self.typ() if self.value is MISSING else self.value
         self.value = self.reducer(base, list(values))

libs/langgraph/tests/test_channels.py

@@ -8,7 +8,6 @@ from langgraph.checkpoint.memory import InMemorySaver
 from langgraph.checkpoint.serde.types import _DeltaSnapshot
 from typing_extensions import NotRequired, TypedDict
 
-from langgraph._internal._constants import OVERWRITE
 from langgraph._internal._typing import MISSING
 from langgraph.channels.binop import BinaryOperatorAggregate
 from langgraph.channels.delta import DeltaChannel
@@ -187,6 +186,18 @@ def test_delta_channel_overwrite() -> None:
     assert ch.get()[0].content == "new"
 
 
+def test_delta_channel_overwrite_bypasses_same_step_reducer_writes() -> None:
+    def list_reducer(state: list, writes: list) -> list:
+        out = list(state)
+        for w in writes:
+            out.extend(w)
+        return out
+
+    ch = DeltaChannel(list_reducer, list).from_checkpoint(MISSING)
+    ch.update([[1], Overwrite([50]), [2]])
+    assert ch.get() == [50]
+
+
 def test_delta_channel_remove_message_and_replay() -> None:
     """RemoveMessage must round-trip correctly when writes are replayed."""
     spec = DeltaChannel(_messages_delta_reducer, list)
@@ -398,49 +409,6 @@ def test_delta_channel_inmemory_saver_assembles_writes() -> None:
     assert len(state.values["messages"]) == 4  # 2 human + 2 AI
 
 
-def test_delta_channel_api_json_overwrite_sentinel_snapshots_and_replays() -> None:
-    def reducer(state: list[str], writes: Sequence[list[str]]) -> list[str]:
-        result = list(state)
-        for write in writes:
-            result.extend(write)
-        return result
-
-    class State(TypedDict):
-        items: Annotated[
-            list[str], DeltaChannel(reducer, list, snapshot_frequency=1000)
-        ]
-
-    calls = 0
-
-    def node(state: State) -> dict:
-        nonlocal calls
-        calls += 1
-        if calls == 1:
-            return {"items": {OVERWRITE: ["reset"]}}
-        return {"items": ["after"]}
-
-    builder = StateGraph(State)
-    builder.add_node("node", node)
-    builder.add_edge(START, "node")
-
-    saver = InMemorySaver()
-    graph = builder.compile(checkpointer=saver)
-    config = {"configurable": {"thread_id": "overwrite-json-sentinel"}}
-
-    updates = list(graph.stream({"items": ["before"]}, config, stream_mode=["updates"]))
-    assert updates == [("updates", {"node": {"items": {OVERWRITE: ["reset"]}}})]
-    assert graph.get_state(config).values == {"items": ["reset"]}
-
-    first_saved = saver.get_tuple(config)
-    assert first_saved is not None
-    snapshot = first_saved.checkpoint["channel_values"].get("items")
-    assert isinstance(snapshot, _DeltaSnapshot)
-    assert snapshot.value == ["reset"]
-
-    assert graph.invoke({"items": []}, config) == {"items": ["reset", "after"]}
-    assert graph.get_state(config).values == {"items": ["reset", "after"]}
-
-
 # ---------------------------------------------------------------------------
 # DeltaChannel — dict reducer
 # ---------------------------------------------------------------------------

libs/langgraph/tests/test_pregel.py

@@ -9281,6 +9281,13 @@ def test_send_with_untracked_value_overlapping_keys(
     assert state.values.get("dictionary") == {"session_resource": "legal_value"}
 
 
+def _delta_list_reducer(state: list, writes: Sequence[list]) -> list:
+    out = list(state)
+    for write in writes:
+        out.extend(write)
+    return out
+
+
 @pytest.mark.parametrize("as_json", [False, True])
 def test_overwrite_sequential(
     sync_checkpointer: BaseCheckpointSaver, as_json: bool
@@ -9388,6 +9395,105 @@ def test_overwrite_parallel_error(
         graph.invoke({"messages": ["START"]}, config)
 
 
+@pytest.mark.parametrize("as_json", [False, True])
+def test_delta_channel_overwrite_sequential(
+    sync_checkpointer: BaseCheckpointSaver, as_json: bool
+) -> None:
+    class State(TypedDict):
+        messages: Annotated[list, DeltaChannel(_delta_list_reducer)]
+
+    def node_a(state: State):
+        return {"messages": ["a"]}
+
+    def node_b(state: State):
+        overwrite = {"__overwrite__": ["b"]} if as_json else Overwrite(["b"])
+        return {"messages": overwrite}
+
+    builder = StateGraph(State)
+    builder.add_node("node_a", node_a)
+    builder.add_node("node_b", node_b)
+    builder.add_edge(START, "node_a")
+    builder.add_edge("node_a", "node_b")
+
+    graph = builder.compile(checkpointer=sync_checkpointer)
+    config = {"configurable": {"thread_id": "delta-overwrite-sequential"}}
+    result = graph.invoke({"messages": ["START"]}, config)
+    assert result == {"messages": ["b"]}
+
+
+@pytest.mark.parametrize("as_json", [False, True])
+def test_delta_channel_overwrite_parallel(
+    sync_checkpointer: BaseCheckpointSaver, as_json: bool
+) -> None:
+    class State(TypedDict):
+        messages: Annotated[list, DeltaChannel(_delta_list_reducer)]
+
+    def node_a(state: State):
+        return {"messages": ["a"]}
+
+    def node_b(state: State):
+        overwrite = {"__overwrite__": ["b"]} if as_json else Overwrite(["b"])
+        return {"messages": overwrite}
+
+    def node_c(state: State):
+        return {"messages": ["c"]}
+
+    def node_d(state: State):
+        return {"messages": ["d"]}
+
+    builder = StateGraph(State)
+    builder.add_node("node_a", node_a)
+    builder.add_node("node_b", node_b)
+    builder.add_node("node_c", node_c)
+    builder.add_node("node_d", node_d)
+    builder.add_edge(START, "node_a")
+    builder.add_edge("node_a", "node_b")
+    builder.add_edge("node_a", "node_c")
+    builder.add_edge("node_b", "node_d")
+    builder.add_edge("node_c", "node_d")
+
+    graph = builder.compile(checkpointer=sync_checkpointer)
+    config = {"configurable": {"thread_id": "delta-overwrite-parallel"}}
+    result = graph.invoke({"messages": ["START"]}, config)
+    assert result == {"messages": ["b", "d"]}
+
+
+@pytest.mark.parametrize("as_json", [False, True])
+def test_delta_channel_overwrite_parallel_error(
+    sync_checkpointer: BaseCheckpointSaver, as_json: bool
+) -> None:
+    class State(TypedDict):
+        messages: Annotated[list, DeltaChannel(_delta_list_reducer)]
+
+    def node_a(state: State):
+        return {"messages": ["a"]}
+
+    def node_b(state: State):
+        overwrite = {"__overwrite__": ["b"]} if as_json else Overwrite(["b"])
+        return {"messages": overwrite}
+
+    def node_c(state: State):
+        overwrite = {"__overwrite__": ["c"]} if as_json else Overwrite(["c"])
+        return {"messages": overwrite}
+
+    builder = StateGraph(State)
+    builder.add_node("node_a", node_a)
+    builder.add_node("node_b", node_b)
+    builder.add_node("node_c", node_c)
+    builder.add_edge(START, "node_a")
+    builder.add_edge("node_a", "node_b")
+    builder.add_edge("node_a", "node_c")
+    builder.add_edge("node_b", END)
+    builder.add_edge("node_c", END)
+
+    graph = builder.compile(checkpointer=sync_checkpointer)
+    config = {"configurable": {"thread_id": "delta-overwrite-parallel-error"}}
+    with pytest.raises(
+        InvalidUpdateError, match="Can receive only one Overwrite value per super-step."
+    ):
+        graph.invoke({"messages": ["START"]}, config)
+
+
 def test_fork_does_not_apply_pending_writes(
     sync_checkpointer: BaseCheckpointSaver,
 ) -> None: