Files
langgraph/libs/checkpoint/tests/test_jsonplus.py
T
5c18bde0f8 feat(langgraph): DeltaChannel: store sentinel in blobs, reconstruct from checkpoint_writes (#7586)
# DeltaChannel: sentinel-based checkpoint blobs + write-replay
reconstruction

## Summary

`DeltaChannel` is a new fold-reducer channel that stores only a
zero-byte sentinel in checkpoint blobs instead of the full accumulated
value. On restore, the runtime replays ancestor writes through the
reducer to reconstruct state. For long-running threads with large
accumulating state (e.g. message histories), this delivers dramatically
smaller checkpoint blobs with configurable read-depth bounds.

```python
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.channels.delta import DeltaChannel
from langgraph.graph.message import _messages_delta_reducer

class State(TypedDict):
    # blob per step: ~60 bytes (sentinel) instead of growing full list
    messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]
    # bound read depth to 10 steps via periodic snapshots
    messages_bounded: Annotated[list, DeltaChannel(_messages_delta_reducer, snapshot_frequency=10)]
```

---

## Storage benchmarks (InMemory, ~400 char/msg)

**Messages blob storage** (`checkpoint_blobs` bytes for the messages
channel):

| turns | add\_messages | delta(inf) | delta(freq=50) | delta(freq=10) |
delta(freq=5) |

|------:|-------------:|-----------:|---------------:|---------------:|--------------:|
| 10 | 91.0 KB | 60 B (1517x) | 60 B (1517x) | 14.4 KB (6x) | 32.6 KB
(3x) |
| 50 | 2.20 MB | 300 B (7347x) | 67.1 KB (33x) | 423 KB (5x) | 864 KB
(3x) |
| 100 | 8.78 MB | 600 B (14636x) | 310 KB (28x) | 1.72 MB (5x) | 3.48 MB
(3x) |
| 250 | 54.80 MB | 1.5 KB (36536x) | 2.09 MB (26x) | 10.87 MB (5x) |
21.84 MB (3x) |
| 500 | 219.19 MB | 3.0 KB (73063x) | 8.56 MB (26x) | 43.67 MB (5x) |
87.50 MB (3x) |

**Total checkpoint storage** (blobs + writes + metadata):

| turns | add\_messages | delta(inf) | delta(freq=50) | delta(freq=10) |
delta(freq=5) |

|------:|-------------:|-----------:|---------------:|---------------:|--------------:|
| 10 | 129.7 KB | 38.7 KB (3.4x) | 38.7 KB (3.4x) | 53.1 KB (2.4x) |
71.2 KB (1.8x) |
| 50 | 2.40 MB | 196 KB (12x) | 263 KB (9x) | 620 KB (3.9x) | 1.06 MB
(2.3x) |
| 100 | 9.18 MB | 394 KB (23x) | 703 KB (13x) | 2.12 MB (4.3x) | 3.87 MB
(2.4x) |
| 250 | 55.79 MB | 987 KB (57x) | 3.07 MB (18x) | 11.86 MB (4.7x) |
22.82 MB (2.4x) |
| 500 | 221.16 MB | 1.98 MB (112x) | 10.53 MB (21x) | 45.64 MB (4.9x) |
89.48 MB (2.5x) |

**Write-phase peak heap**:

| turns | add\_messages | delta(inf) | delta(freq=50) | delta(freq=10) |
delta(freq=5) |

|------:|-------------:|-----------:|---------------:|---------------:|--------------:|
| 10 | 456 KB | 199 KB (2.3x) | 199 KB (2.3x) | 212 KB (2.2x) | 232 KB
(2.0x) |
| 50 | 3.04 MB | 742 KB (4.1x) | 805 KB (3.8x) | 1.21 MB (2.5x) | 1.67
MB (1.8x) |
| 100 | 10.70 MB | 1.41 MB (7.6x) | 1.82 MB (5.9x) | 3.42 MB (3.1x) |
5.25 MB (2.0x) |
| 250 | 60.44 MB | 3.36 MB (18x) | 5.67 MB (11x) | 14.87 MB (4.1x) |
26.31 MB (2.3x) |

**Read-phase avg `get_state` latency** (5 calls, InMemory):

| turns | add\_messages | delta(inf) | delta(freq=50) | delta(freq=10) |
delta(freq=5) |

|------:|-------------:|-----------:|---------------:|---------------:|--------------:|
| 10 | 0.7 ms | 1.1 ms (0.6x) | 1.1 ms (0.6x) | 0.8 ms (0.9x) | 0.6 ms
(1.1x) |
| 50 | 2.7 ms | 5.3 ms (0.5x) | 3.5 ms (0.8x) | 2.7 ms (1.0x) | 2.7 ms
(1.0x) |
| 100 | 5.5 ms | 11.1 ms (0.5x) | 6.0 ms (0.9x) | 5.2 ms (1.1x) | 5.4 ms
(1.0x) |
| 250 | 12.9 ms | 27.2 ms (0.5x) | 13.6 ms (0.9x) | 12.9 ms (1.0x) |
13.0 ms (1.0x) |

**Postgres `get_tuple` read latency** (~100 tok/msg per step):

| steps | full-list | delta(inf) | delta(freq=50) | delta(freq=10) |
delta(freq=5) |

|------:|----------:|-----------:|---------------:|---------------:|--------------:|
| 10 | 0.29 ms | 0.21 ms (1.4x) | 0.19 ms (1.6x) | 0.19 ms (1.5x) | 0.19
ms (1.6x) |
| 50 | 0.19 ms | 0.15 ms (1.3x) | 0.19 ms (1.0x) | 0.22 ms (0.8x) | 0.29
ms (0.7x) |
| 100 | 0.27 ms | 0.17 ms (1.6x) | 0.22 ms (1.2x) | 0.23 ms (1.2x) |
0.21 ms (1.3x) |
| 500 | 0.60 ms | 0.30 ms (2.0x) | 0.66 ms (0.9x) | 0.56 ms (1.1x) |
0.69 ms (0.9x) |

**Takeaway:** `snapshot_frequency=10` matches full-list read latency
while still saving 5x on blob storage and ~4x on total storage.

---

## How it works

### Checkpoint blobs

`checkpoint()` always returns `DELTA_SENTINEL` (a zero-byte msgpack ext
marker) instead of the accumulated value. On restore, the saver's
`_get_channel_writes_history` walks the ancestor chain collecting
`checkpoint_writes` entries and replays them through the reducer:

```python
# blob stored per step: ~1 byte (sentinel)
# vs. full list growing O(N) every step with BinaryOperatorAggregate
```

### Reducer interface

`DeltaChannel` takes a **batch reducer** `(state, list[writes]) ->
state` — all writes for a step arrive in one call, enabling single-pass
implementations:

```python
#  Don't use add_messages directly — it's a binary operator, not a batch reducer
messages: Annotated[list, DeltaChannel(add_messages)]  # wrong

#  Use _messages_delta_reducer — single pass, dedup by ID, RemoveMessage support
messages: Annotated[list, DeltaChannel(_messages_delta_reducer)]

#  Or write your own batch reducer for custom types
def my_dict_reducer(state: dict, writes: list[dict]) -> dict:
    result = dict(state)
    for w in writes:
        result.update(w)
    return result

files: Annotated[dict, DeltaChannel(my_dict_reducer)]
```

### Snapshot frequency

`snapshot_frequency=N` writes a full `_DeltaSnapshot` blob every N
pregel steps, bounding replay depth regardless of thread length.
Snapshots are eager — written even if the channel had no update that
step, so the depth bound always holds:

```python
# Replay walks at most 10 ancestors before hitting a snapshot
messages: Annotated[list, DeltaChannel(_messages_delta_reducer, snapshot_frequency=10)]
```

### Migration from `BinaryOperatorAggregate`

Pre-existing threads written under `BinaryOperatorAggregate` work
transparently after swapping the annotation — the saver detects a
plain-value ancestor blob and uses it as the reconstruction seed:

```python
# Before: BinaryOperatorAggregate stores full list every step
items: Annotated[list, add_messages]

# After: DeltaChannel — existing checkpoints still readable, new steps use sentinel
items: Annotated[list, DeltaChannel(_messages_delta_reducer)]
```

### Async write-ordering safety

In `durability="async"` mode (default), `put_writes` calls are
fire-and-forget. `AsyncPregelLoop` tracks in-flight `aput_writes`
futures for DeltaChannel channels in `_delta_write_futs` and drains them
via `await asyncio.gather()` in `_checkpointer_put_after_previous`
before `aput()` — ensuring `checkpoint_writes` are durable before the
sentinel blob is committed.

---

## What's in scope

- **`libs/langgraph/langgraph/channels/delta.py`** — `DeltaChannel`
implementation
- **`libs/langgraph/langgraph/graph/message.py`** —
`_messages_delta_reducer` (experimental)
- **`libs/checkpoint/`** — `_get_channel_writes_history` ancestor-walk
API on `BaseCheckpointSaver`, `InMemorySaver` optimized override
- **`libs/checkpoint-postgres/`** — `PostgresSaver` /
`AsyncPostgresSaver` single-roundtrip UNION ALL override
- **`libs/langgraph/langgraph/pregel/`** — `channels_from_checkpoint` /
`create_checkpoint` wiring, async write-ordering safety

---

## Follow-ups

- **Batch reconstruction**: each DeltaChannel field issues its own
`_get_channel_writes_history` call; a single walk collecting all
sentinel channels would reduce roundtrips proportionally to the number
of DeltaChannel fields.
- **Sync write ordering**: `BackgroundExecutor.__exit__` guarantees
completion before `invoke()` returns, but within a run there's no
explicit ordering between `put_writes` and `put`. Two-phase commit for
sync would close this gap.
- **`ShallowPostgresSaver` compatibility**: shallow savers keep only the
latest checkpoint and have no parent chain to walk; DeltaChannel is
currently incompatible and should raise or warn at compile time.
- Updating the writes table w/ delta epoch ids for more efficient reads
- follow up w/ LSD checkpointer implementations to support delta
channel! and update prune

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: ccurme <chester.curme@gmail.com>
Co-authored-by: Will Fu-Hinthorn <will@langchain.dev>
2026-04-29 17:26:17 -04:00

1063 lines
33 KiB
Python

import dataclasses
import json
import logging
import pathlib
import re
import sys
import uuid
from collections import deque
from datetime import date, datetime, time, timezone
from decimal import Decimal
from enum import Enum
from ipaddress import IPv4Address
from zoneinfo import ZoneInfo
import dataclasses_json
import numpy as np
import ormsgpack
import pandas as pd
import pytest
from langchain_core.documents.base import Document
from langchain_core.messages import HumanMessage
from pydantic import BaseModel, SecretStr
from pydantic.v1 import BaseModel as BaseModelV1
from pydantic.v1 import SecretStr as SecretStrV1
from langgraph.checkpoint.serde import _msgpack as _lg_msgpack
from langgraph.checkpoint.serde._msgpack import AllowedMsgpackModules
from langgraph.checkpoint.serde.event_hooks import (
SerdeEvent,
register_serde_event_listener,
)
from langgraph.checkpoint.serde.jsonplus import (
EXT_METHOD_SINGLE_ARG,
InvalidModuleError,
JsonPlusSerializer,
_msgpack_enc,
_msgpack_ext_hook_to_json,
_warned_blocked_types,
_warned_unregistered_types,
)
from langgraph.store.base import Item
class InnerPydantic(BaseModel):
hello: str
class MyPydantic(BaseModel):
foo: str
bar: int
inner: InnerPydantic
class AnotherPydantic(BaseModel):
foo: str
class InnerPydanticV1(BaseModelV1):
hello: str
class MyPydanticV1(BaseModelV1):
foo: str
bar: int
inner: InnerPydanticV1
@dataclasses.dataclass
class InnerDataclass:
hello: str
@dataclasses.dataclass
class MyDataclass:
foo: str
bar: int
inner: InnerDataclass
def something(self) -> None:
pass
@dataclasses.dataclass(slots=True)
class MyDataclassWSlots:
foo: str
bar: int
inner: InnerDataclass
def something(self) -> None:
pass
class MyEnum(Enum):
FOO = "foo"
BAR = "bar"
@dataclasses_json.dataclass_json
@dataclasses.dataclass
class Person:
name: str
def test_serde_jsonplus() -> None:
uid = uuid.UUID(int=1)
deque_instance = deque([1, 2, 3])
tzn = ZoneInfo("America/New_York")
ip4 = IPv4Address("192.168.0.1")
current_date = date(2024, 4, 19)
current_time = time(23, 4, 57, 51022, timezone.max)
current_timestamp = datetime(2024, 4, 19, 23, 4, 57, 51022, timezone.max)
to_serialize = {
"path": pathlib.Path("foo", "bar"),
"re": re.compile(r"foo", re.DOTALL),
"decimal": Decimal("1.10101"),
"set": {1, 2, frozenset({1, 2})},
"frozen_set": frozenset({1, 2, 3}),
"ip4": ip4,
"deque": deque_instance,
"tzn": tzn,
"date": current_date,
"time": current_time,
"uid": uid,
"timestamp": current_timestamp,
"my_rich_dict": {(1, 2, 3): 45},
"my_slotted_class": MyDataclassWSlots("bar", 2, InnerDataclass("hello")),
"my_dataclass": MyDataclass("foo", 1, InnerDataclass("hello")),
"my_enum": MyEnum.FOO,
"my_pydantic": MyPydantic(foo="foo", bar=1, inner=InnerPydantic(hello="hello")),
"my_secret_str": SecretStr("meow"),
"person": Person(name="foo"),
"a_bool": True,
"a_none": None,
"a_str": "foo",
"a_str_nuc": "foo\u0000",
"a_str_uc": "foo ⛰️",
"a_str_ucuc": "foo \u26f0\ufe0f\u0000",
"a_str_ucucuc": "foo \\u26f0\\ufe0f",
"an_int": 1,
"a_float": 1.1,
"a_bytes": b"my bytes",
"a_bytearray": bytearray([42]),
"my_item": Item(
value={},
key="my-key",
namespace=("a", "name", " "),
created_at=datetime(2024, 9, 24, 17, 29, 10, 128397),
updated_at=datetime(2024, 9, 24, 17, 29, 11, 128397),
),
}
if sys.version_info < (3, 14):
to_serialize["my_pydantic_v1"] = MyPydanticV1(
foo="foo", bar=1, inner=InnerPydanticV1(hello="hello")
)
to_serialize["my_secret_str_v1"] = SecretStrV1("meow")
allowed_msgpack_modules: AllowedMsgpackModules = [
InnerDataclass,
MyDataclass,
MyDataclassWSlots,
MyEnum,
InnerPydantic,
MyPydantic,
# Testing that it supports both.
(Person.__module__, Person.__name__),
(SecretStr.__module__, SecretStr.__name__),
]
if sys.version_info < (3, 14):
allowed_msgpack_modules.extend( # type: ignore
[
(InnerPydanticV1.__module__, InnerPydanticV1.__name__),
(MyPydanticV1.__module__, MyPydanticV1.__name__),
(SecretStrV1.__module__, SecretStrV1.__name__),
]
)
serde = JsonPlusSerializer(allowed_msgpack_modules=allowed_msgpack_modules)
dumped = serde.dumps_typed(to_serialize)
assert dumped[0] == "msgpack"
assert serde.loads_typed(dumped) == to_serialize
for value in to_serialize.values():
assert serde.loads_typed(serde.dumps_typed(value)) == value
surrogates = [
"Hello??",
"Python??",
"Surrogate??",
"Example??",
"String??",
"With??",
"Surrogates??",
"Embedded??",
"In??",
"The??",
"Text??",
"收花🙄·到",
]
serde = JsonPlusSerializer(pickle_fallback=False)
assert serde.loads_typed(serde.dumps_typed(surrogates)) == surrogates
def test_serde_jsonplus_json_mode() -> None:
uid = uuid.UUID(int=1)
deque_instance = deque([1, 2, 3])
tzn = ZoneInfo("America/New_York")
ip4 = IPv4Address("192.168.0.1")
current_date = date(2024, 4, 19)
current_time = time(23, 4, 57, 51022, timezone.max)
current_timestamp = datetime(2024, 4, 19, 23, 4, 57, 51022, timezone.max)
to_serialize = {
"path": pathlib.Path("foo", "bar"),
"re": re.compile(r"foo", re.DOTALL),
"decimal": Decimal("1.10101"),
"set": {1, 2, frozenset({1, 2})},
"frozen_set": frozenset({1, 2, 3}),
"ip4": ip4,
"deque": deque_instance,
"tzn": tzn,
"date": current_date,
"time": current_time,
"uid": uid,
"timestamp": current_timestamp,
"my_slotted_class": MyDataclassWSlots("bar", 2, InnerDataclass("hello")),
"my_dataclass": MyDataclass("foo", 1, InnerDataclass("hello")),
"my_enum": MyEnum.FOO,
"my_pydantic": MyPydantic(foo="foo", bar=1, inner=InnerPydantic(hello="hello")),
"my_secret_str": SecretStr("meow"),
"person": Person(name="foo"),
"a_bool": True,
"a_none": None,
"a_str": "foo",
"a_str_nuc": "foo\u0000",
"a_str_uc": "foo ⛰️",
"a_str_ucuc": "foo \u26f0\ufe0f\u0000",
"a_str_ucucuc": "foo \\u26f0\\ufe0f",
"an_int": 1,
"a_float": 1.1,
"a_bytes": b"my bytes",
"a_bytearray": bytearray([42]),
"my_item": Item(
value={},
key="my-key",
namespace=("a", "name", " "),
created_at=datetime(2024, 9, 24, 17, 29, 10, 128397),
updated_at=datetime(2024, 9, 24, 17, 29, 11, 128397),
),
}
if sys.version_info < (3, 14):
to_serialize["my_pydantic_v1"] = MyPydanticV1(
foo="foo", bar=1, inner=InnerPydanticV1(hello="hello")
)
to_serialize["my_secret_str_v1"] = SecretStrV1("meow")
serde = JsonPlusSerializer(__unpack_ext_hook__=_msgpack_ext_hook_to_json)
dumped = serde.dumps_typed(to_serialize)
assert dumped[0] == "msgpack"
result = serde.loads_typed(dumped)
expected_result = {
"path": ["foo", "bar"],
"re": ["foo", 48],
"decimal": "1.10101",
"set": [1, 2, [1, 2]],
"frozen_set": [1, 2, 3],
"ip4": "192.168.0.1",
"deque": [1, 2, 3],
"tzn": "America/New_York",
"date": [2024, 4, 19],
"time": {
"hour": 23,
"minute": 4,
"second": 57,
"microsecond": 51022,
"tzinfo": [[0, 86340, 0]],
"fold": 0,
},
"uid": "00000000-0000-0000-0000-000000000001",
"timestamp": "2024-04-19T23:04:57.051022+23:59",
"my_slotted_class": {"foo": "bar", "bar": 2, "inner": {"hello": "hello"}},
"my_dataclass": {"foo": "foo", "bar": 1, "inner": {"hello": "hello"}},
"my_enum": "foo",
"my_pydantic": {"foo": "foo", "bar": 1, "inner": {"hello": "hello"}},
"my_secret_str": "meow",
"person": {"name": "foo"},
"a_bool": True,
"a_none": None,
"a_str": "foo",
"a_str_nuc": "foo\x00",
"a_str_uc": "foo ⛰️",
"a_str_ucuc": "foo ⛰️\x00",
"a_str_ucucuc": "foo \\u26f0\\ufe0f",
"an_int": 1,
"a_float": 1.1,
"a_bytes": b"my bytes",
"a_bytearray": b"*",
"my_item": {
"namespace": ["a", "name", " "],
"key": "my-key",
"value": {},
"created_at": "2024-09-24T17:29:10.128397",
"updated_at": "2024-09-24T17:29:11.128397",
},
}
if sys.version_info < (3, 14):
expected_result["my_pydantic_v1"] = {
"foo": "foo",
"bar": 1,
"inner": {"hello": "hello"},
}
expected_result["my_secret_str_v1"] = "meow"
assert result == expected_result
def test_serde_jsonplus_bytes() -> None:
serde = JsonPlusSerializer()
some_bytes = b"my bytes"
dumped = serde.dumps_typed(some_bytes)
assert dumped == ("bytes", some_bytes)
assert serde.loads_typed(dumped) == some_bytes
def test_lc2_json_safe_type_revives_without_allowlist() -> None:
"""Old 'json' blobs with lc=2 for safe types must revive without an explicit allowlist.
Regression test for: https://github.com/langchain-ai/langgraph/issues/7498
Threads checkpointed before v1.0.1 (pre-msgpack) stored messages as lc=2 JSON
constructor dicts. Resuming those threads must reconstruct proper BaseMessage objects
rather than returning raw dicts that cause MESSAGE_COERCION_FAILURE in add_messages.
"""
from langchain_core.messages import AIMessage
serde = JsonPlusSerializer() # default: _allowed_json_modules=None
human_blob = {
"lc": 2,
"type": "constructor",
"id": ["langchain_core", "messages", "human", "HumanMessage"],
"kwargs": {"content": "hello", "type": "human"},
}
ai_blob = {
"lc": 2,
"type": "constructor",
"id": ["langchain_core", "messages", "ai", "AIMessage"],
"kwargs": {"content": "hi there", "type": "ai"},
}
result = serde.loads_typed(("json", json.dumps([human_blob, ai_blob]).encode()))
assert len(result) == 2
assert isinstance(result[0], HumanMessage), (
f"Expected HumanMessage, got {type(result[0])}: {result[0]!r}\n"
"lc=2 JSON blobs for safe types must deserialize without an explicit allowlist"
)
assert result[0].content == "hello"
assert isinstance(result[1], AIMessage)
assert result[1].content == "hi there"
def test_lc2_json_unknown_type_stays_blocked_without_allowlist() -> None:
"""lc=2 JSON blobs for types NOT in SAFE_MSGPACK_TYPES still require an allowlist."""
serde = JsonPlusSerializer()
load = {
"lc": 2,
"type": "constructor",
"id": ["pprint", "pprint"],
"kwargs": {"object": "HELLO"},
}
# No allowlist configured → raw dict returned (not raised, not reconstructed)
result = serde.loads_typed(("json", json.dumps(load).encode()))
assert isinstance(result, dict), "Unknown lc=2 type must stay as raw dict"
assert result.get("lc") == 2
def test_deserde_invalid_module() -> None:
serde = JsonPlusSerializer()
load = {
"lc": 2,
"type": "constructor",
"id": ["pprint", "pprint"],
"kwargs": {"object": "HELLO"},
}
with pytest.raises(InvalidModuleError):
serde._revive_lc2(load)
serde = JsonPlusSerializer(allowed_json_modules=[("pprint", "pprint")])
serde.loads_typed(("json", json.dumps(load).encode("utf-8")))
def test_serde_jsonplus_bytearray() -> None:
serde = JsonPlusSerializer()
some_bytearray = bytearray([42])
dumped = serde.dumps_typed(some_bytearray)
assert dumped == ("bytearray", some_bytearray)
assert serde.loads_typed(dumped) == some_bytearray
@pytest.mark.parametrize(
"arr",
[
np.arange(9, dtype=np.int32).reshape(3, 3),
np.asfortranarray(np.arange(9, dtype=np.float64).reshape(3, 3)),
np.arange(12, dtype=np.int16)[::2].reshape(3, 2),
],
)
def test_serde_jsonplus_numpy_array(arr: np.ndarray) -> None:
serde = JsonPlusSerializer()
dumped = serde.dumps_typed(arr)
assert dumped[0] == "msgpack"
result = serde.loads_typed(dumped)
assert isinstance(result, np.ndarray)
assert result.dtype == arr.dtype
assert np.array_equal(result, arr)
@pytest.mark.parametrize(
"arr",
[
np.arange(6, dtype=np.float32).reshape(2, 3),
np.asfortranarray(np.arange(4, dtype=np.complex128).reshape(2, 2)),
],
)
def test_serde_jsonplus_numpy_array_json_hook(arr: np.ndarray) -> None:
serde = JsonPlusSerializer(__unpack_ext_hook__=_msgpack_ext_hook_to_json)
dumped = serde.dumps_typed(arr)
assert dumped[0] == "msgpack"
result = serde.loads_typed(dumped)
assert isinstance(result, list)
assert result == arr.tolist()
@pytest.mark.parametrize(
"df",
[
pd.DataFrame(),
pd.DataFrame({"int_col": [1, 2, 3]}),
pd.DataFrame({"float_col": [1.1, 2.2, 3.3]}),
pd.DataFrame({"str_col": ["a", "b", "c"]}),
pd.DataFrame({"bool_col": [True, False, True]}),
pd.DataFrame(
{
"datetime_col": [
datetime(2024, 1, 1),
datetime(2024, 1, 2),
datetime(2024, 1, 3),
]
}
),
pd.DataFrame(
{
"int_col": [1, 2, 3],
"float_col": [1.1, 2.2, 3.3],
"str_col": ["a", "b", "c"],
}
),
pd.DataFrame(
{
"int_col": [1, 2, None],
"float_col": [1.1, None, 3.3],
"str_col": ["a", None, "c"],
}
),
pytest.param(
pd.DataFrame({"cat_col": pd.Categorical(["a", "b", "a", "c"])}),
marks=pytest.mark.skipif(
sys.version_info >= (3, 14), reason="NotImplementedError on Python 3.14"
),
),
pd.DataFrame(
{
"int8": pd.array([1, 2, 3], dtype="int8"),
"int16": pd.array([10, 20, 30], dtype="int16"),
"int32": pd.array([100, 200, 300], dtype="int32"),
"int64": pd.array([1000, 2000, 3000], dtype="int64"),
"float32": pd.array([1.1, 2.2, 3.3], dtype="float32"),
"float64": pd.array([10.1, 20.2, 30.3], dtype="float64"),
}
),
pd.DataFrame({"value": [1, 2, 3]}, index=["x", "y", "z"]),
pd.DataFrame(
[[1, 2, 3, 4]],
columns=pd.MultiIndex.from_tuples(
[("A", "X"), ("A", "Y"), ("B", "X"), ("B", "Y")]
),
),
pd.DataFrame(
{"value": [1, 2, 3]}, index=pd.date_range("2024-01-01", periods=3, freq="D")
),
pd.DataFrame(
{
"col1": range(1000),
"col2": [f"str_{i}" for i in range(1000)],
"col3": np.random.rand(1000),
}
),
pytest.param(
pd.DataFrame(
{
"tz_datetime": pd.date_range(
"2024-01-01", periods=3, freq="D", tz="UTC"
)
}
),
marks=pytest.mark.skipif(
sys.version_info >= (3, 14), reason="NotImplementedError on Python 3.14"
),
),
pd.DataFrame({"timedelta": pd.to_timedelta([1, 2, 3], unit="D")}),
pytest.param(
pd.DataFrame({"period": pd.period_range("2024-01", periods=3, freq="M")}),
marks=pytest.mark.skipif(
sys.version_info >= (3, 14), reason="NotImplementedError on Python 3.14"
),
),
pd.DataFrame({"interval": pd.interval_range(start=0, end=3, periods=3)}),
pd.DataFrame({"unicode": ["Hello 🌍", "Python 🐍", "Data 📊"]}),
pd.DataFrame({"mixed": [1, "string", [1, 2, 3], {"key": "value"}]}),
pd.DataFrame({"a": [1], "b": ["test"], "c": [3.14]}),
pd.DataFrame({"single": [42]}),
pd.DataFrame(
{
"small": [sys.float_info.min, 0, sys.float_info.max],
"large_int": [-(2**63), 0, 2**63 - 1],
}
),
pd.DataFrame({"special_strings": ["", "null", "None", "NaN", "inf", "-inf"]}),
pd.DataFrame({"bytes_col": [b"hello", b"world", b"\x00\x01\x02"]}),
],
)
def test_serde_jsonplus_pandas_dataframe(df: pd.DataFrame) -> None:
serde = JsonPlusSerializer(pickle_fallback=True)
dumped = serde.dumps_typed(df)
assert dumped[0] == "pickle"
result = serde.loads_typed(dumped)
assert result.equals(df)
@pytest.mark.parametrize(
"series",
[
pd.Series([]),
pd.Series([1, 2, 3]),
pd.Series([1.1, 2.2, 3.3]),
pd.Series(["a", "b", "c"]),
pd.Series([True, False, True]),
pd.Series([datetime(2024, 1, 1), datetime(2024, 1, 2), datetime(2024, 1, 3)]),
pd.Series([1, 2, None]),
pd.Series([1.1, None, 3.3]),
pd.Series(["a", None, "c"]),
pytest.param(
pd.Series(pd.Categorical(["a", "b", "a", "c"])),
marks=pytest.mark.skipif(
sys.version_info >= (3, 14), reason="NotImplementedError on Python 3.14"
),
),
pd.Series([1, 2, 3], dtype="int8"),
pd.Series([10, 20, 30], dtype="int16"),
pd.Series([100, 200, 300], dtype="int32"),
pd.Series([1000, 2000, 3000], dtype="int64"),
pd.Series([1.1, 2.2, 3.3], dtype="float32"),
pd.Series([10.1, 20.2, 30.3], dtype="float64"),
pd.Series([1, 2, 3], index=["x", "y", "z"]),
pd.Series([1, 2, 3], index=pd.date_range("2024-01-01", periods=3, freq="D")),
pd.Series(range(1000)),
pd.Series(pd.date_range("2024-01-01", periods=3, freq="D", tz="UTC")),
pd.Series(pd.to_timedelta([1, 2, 3], unit="D")),
pd.Series(pd.period_range("2024-01", periods=3, freq="M")),
pd.Series(pd.interval_range(start=0, end=3, periods=3)),
pd.Series(["Hello 🌍", "Python 🐍", "Data 📊"]),
pd.Series([1, "string", [1, 2, 3], {"key": "value"}]),
pd.Series([42], name="single"),
pd.Series([sys.float_info.min, 0, sys.float_info.max]),
pd.Series([-(2**63), 0, 2**63 - 1]),
pd.Series(["", "null", "None", "NaN", "inf", "-inf"]),
pd.Series([b"hello", b"world", b"\x00\x01\x02"]),
pd.Series([1, 2, 3], name="named_series"),
pd.Series(
[10, 20],
index=pd.MultiIndex.from_tuples([("a", 1), ("b", 2)], names=["x", "y"]),
),
],
)
def test_serde_jsonplus_pandas_series(series: pd.Series) -> None:
serde = JsonPlusSerializer(pickle_fallback=True)
dumped = serde.dumps_typed(series)
assert dumped[0] == "pickle"
result = serde.loads_typed(dumped)
assert result.equals(series)
def test_msgpack_safe_types_no_warning(caplog: pytest.LogCaptureFixture) -> None:
"""Test safe types deserialize without warnings."""
serde = JsonPlusSerializer()
safe_objects = [
datetime.now(),
date.today(),
time(12, 30),
timezone.utc,
uuid.uuid4(),
Decimal("123.45"),
{1, 2, 3},
frozenset([1, 2, 3]),
deque([1, 2, 3]),
IPv4Address("192.168.1.1"),
pathlib.Path("/tmp/test"),
]
for obj in safe_objects:
caplog.clear()
dumped = serde.dumps_typed(obj)
result = serde.loads_typed(dumped)
assert "unregistered type" not in caplog.text.lower(), (
f"Unexpected warning for {type(obj)}"
)
assert result is not None
@pytest.fixture(autouse=True)
def _reset_warned_types() -> None:
# Warning dedup state is process-global; reset per-test so each case sees
# a fresh slate and assertions about warning emission are stable.
_warned_unregistered_types.clear()
_warned_blocked_types.clear()
def test_msgpack_pydantic_warns_by_default(caplog: pytest.LogCaptureFixture) -> None:
"""Pydantic models not in allowlist should log warning but still deserialize."""
current = _lg_msgpack.STRICT_MSGPACK_ENABLED
_lg_msgpack.STRICT_MSGPACK_ENABLED = False
serde = JsonPlusSerializer()
obj = MyPydantic(foo="test", bar=42, inner=InnerPydantic(hello="world"))
caplog.clear()
dumped = serde.dumps_typed(obj)
result = serde.loads_typed(dumped)
assert "unregistered type" in caplog.text.lower()
assert "allowed_msgpack_modules" in caplog.text
assert result == obj
# Second deserialization of the same type should NOT produce another warning
caplog.clear()
result2 = serde.loads_typed(dumped)
assert "unregistered type" not in caplog.text.lower()
assert result2 == obj
_lg_msgpack.STRICT_MSGPACK_ENABLED = current
def test_msgpack_env_strict_default(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Strict msgpack env should default to blocking unregistered types."""
current = _lg_msgpack.STRICT_MSGPACK_ENABLED
_lg_msgpack.STRICT_MSGPACK_ENABLED = True
serde = JsonPlusSerializer()
obj = MyPydantic(foo="test", bar=42, inner=InnerPydantic(hello="world"))
caplog.clear()
dumped = serde.dumps_typed(obj)
result = serde.loads_typed(dumped)
assert "blocked" in caplog.text.lower()
assert result == obj.model_dump()
_lg_msgpack.STRICT_MSGPACK_ENABLED = current
def test_msgpack_allowlist_silences_warning(caplog: pytest.LogCaptureFixture) -> None:
"""Types in allowed_msgpack_modules should deserialize without warnings."""
serde = JsonPlusSerializer(
allowed_msgpack_modules=[
("tests.test_jsonplus", "MyPydantic"),
("tests.test_jsonplus", "InnerPydantic"),
]
)
obj = MyPydantic(foo="test", bar=42, inner=InnerPydantic(hello="world"))
caplog.clear()
dumped = serde.dumps_typed(obj)
result = serde.loads_typed(dumped)
assert "unregistered type" not in caplog.text.lower()
assert result == obj
def test_msgpack_none_blocks_unregistered(caplog: pytest.LogCaptureFixture) -> None:
"""allowed_msgpack_modules=None should block unregistered types."""
serde = JsonPlusSerializer(allowed_msgpack_modules=None)
obj = MyPydantic(foo="test", bar=42, inner=InnerPydantic(hello="world"))
caplog.clear()
dumped = serde.dumps_typed(obj)
result = serde.loads_typed(dumped)
assert "blocked" in caplog.text.lower()
expected = obj.model_dump()
assert result == expected
def test_msgpack_allowlist_blocks_non_listed(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Allowlists should block unregistered types even if msgpack is enabled."""
serde = JsonPlusSerializer(
allowed_msgpack_modules=[("tests.test_jsonplus", "MyPydantic")]
)
obj = AnotherPydantic(foo="nope")
caplog.clear()
dumped = serde.dumps_typed(obj)
result = serde.loads_typed(dumped)
assert "blocked" in caplog.text.lower()
expected = obj.model_dump()
# It's not allowed, so we just leave it as a dict
assert result == expected
def test_msgpack_blocked_emits_event() -> None:
events: list[SerdeEvent] = []
unregister = register_serde_event_listener(events.append)
try:
serde = JsonPlusSerializer(allowed_msgpack_modules=None)
obj = AnotherPydantic(foo="nope")
serde.loads_typed(serde.dumps_typed(obj))
finally:
unregister()
assert {
"kind": "msgpack_blocked",
"module": "tests.test_jsonplus",
"name": "AnotherPydantic",
} in events
def test_msgpack_unregistered_allowed_emits_event() -> None:
events: list[SerdeEvent] = []
unregister = register_serde_event_listener(events.append)
try:
serde = JsonPlusSerializer(allowed_msgpack_modules=True)
obj = AnotherPydantic(foo="ok")
serde.loads_typed(serde.dumps_typed(obj))
finally:
unregister()
assert {
"kind": "msgpack_unregistered_allowed",
"module": "tests.test_jsonplus",
"name": "AnotherPydantic",
} in events
def test_msgpack_strict_allows_safe_types(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Safe types should still deserialize in strict mode without warnings."""
serde = JsonPlusSerializer(allowed_msgpack_modules=None)
safe = uuid.uuid4()
caplog.clear()
dumped = serde.dumps_typed(safe)
result = serde.loads_typed(dumped)
assert "blocked" not in caplog.text.lower()
assert result == safe
def test_msgpack_strict_allows_core_langchain_messages(
caplog: pytest.LogCaptureFixture,
) -> None:
serde = JsonPlusSerializer(allowed_msgpack_modules=None)
msg = HumanMessage(content="hello")
caplog.clear()
result = serde.loads_typed(serde.dumps_typed(msg))
assert "blocked" not in caplog.text.lower()
assert "unregistered" not in caplog.text.lower()
assert isinstance(result, HumanMessage)
assert result == msg
def test_msgpack_strict_allows_langchain_document(
caplog: pytest.LogCaptureFixture,
) -> None:
serde = JsonPlusSerializer(allowed_msgpack_modules=None)
doc = Document(page_content="hello", metadata={"k": "v"})
caplog.clear()
result = serde.loads_typed(serde.dumps_typed(doc))
assert "blocked" not in caplog.text.lower()
assert "unregistered" not in caplog.text.lower()
assert isinstance(result, Document)
assert result == doc
def test_msgpack_regex_safe_type(caplog: pytest.LogCaptureFixture) -> None:
"""re.compile patterns should deserialize without warnings as a safe type."""
serde = JsonPlusSerializer(allowed_msgpack_modules=None)
pattern = re.compile(r"foo.*bar", re.IGNORECASE | re.DOTALL)
caplog.clear()
dumped = serde.dumps_typed(pattern)
result = serde.loads_typed(dumped)
assert "blocked" not in caplog.text.lower()
assert "unregistered" not in caplog.text.lower()
assert result.pattern == pattern.pattern
assert result.flags == pattern.flags
def test_msgpack_method_pathlib_blocked_in_strict(
tmp_path: pathlib.Path, caplog: pytest.LogCaptureFixture
) -> None:
target = tmp_path / "secret.txt"
target.write_text("secret")
serde = JsonPlusSerializer(allowed_msgpack_modules=None)
payload = ormsgpack.packb(
ormsgpack.Ext(
EXT_METHOD_SINGLE_ARG,
_msgpack_enc(("pathlib", "Path", target, "read_text")),
),
option=ormsgpack.OPT_NON_STR_KEYS,
)
caplog.set_level(logging.WARNING, logger="langgraph.checkpoint.serde.jsonplus")
caplog.clear()
result = serde.loads_typed(("msgpack", payload))
assert result == target
assert "blocked deserialization of method call pathlib.path.read_text" in (
caplog.text.lower()
)
def test_msgpack_method_pathlib_blocked_default_mode(
tmp_path: pathlib.Path, caplog: pytest.LogCaptureFixture
) -> None:
target = tmp_path / "secret.txt"
target.write_text("secret")
serde = JsonPlusSerializer(allowed_msgpack_modules=True)
payload = ormsgpack.packb(
ormsgpack.Ext(
EXT_METHOD_SINGLE_ARG,
_msgpack_enc(("pathlib", "Path", target, "read_text")),
),
option=ormsgpack.OPT_NON_STR_KEYS,
)
caplog.set_level(logging.WARNING, logger="langgraph.checkpoint.serde.jsonplus")
caplog.clear()
result = serde.loads_typed(("msgpack", payload))
assert result == target
assert "blocked deserialization of method call pathlib.path.read_text" in (
caplog.text.lower()
)
def test_msgpack_regex_still_works_strict(caplog: pytest.LogCaptureFixture) -> None:
serde = JsonPlusSerializer(allowed_msgpack_modules=None)
pattern = re.compile(r"pattern", re.IGNORECASE | re.MULTILINE)
caplog.clear()
result = serde.loads_typed(serde.dumps_typed(pattern))
assert "blocked" not in caplog.text.lower()
assert result.pattern == pattern.pattern
assert result.flags == pattern.flags
def test_msgpack_path_constructor_still_works() -> None:
serde = JsonPlusSerializer(allowed_msgpack_modules=None)
path_obj = pathlib.Path("/tmp/foo")
result = serde.loads_typed(serde.dumps_typed(path_obj))
assert result == path_obj
def test_with_msgpack_allowlist_noop_returns_same_instance() -> None:
serde = JsonPlusSerializer(allowed_msgpack_modules=None)
result = serde.with_msgpack_allowlist(())
assert result is serde
def test_with_msgpack_allowlist_supports_subclass_without_init_kwargs() -> None:
class CustomSerializer(JsonPlusSerializer):
def __init__(self) -> None:
super().__init__(allowed_msgpack_modules=None)
serde = CustomSerializer()
result = serde.with_msgpack_allowlist([MyDataclass])
assert isinstance(result, CustomSerializer)
assert result is not serde
assert serde._allowed_msgpack_modules is None
assert result._allowed_msgpack_modules == {
(MyDataclass.__module__, MyDataclass.__name__)
}
def test_with_msgpack_allowlist_rebuilds_default_unpack_hook() -> None:
serde = JsonPlusSerializer(allowed_msgpack_modules=None)
original_hook = serde._unpack_ext_hook
result = serde.with_msgpack_allowlist([MyDataclass])
assert result._unpack_ext_hook is not original_hook
def test_with_msgpack_allowlist_preserves_custom_unpack_hook() -> None:
def custom_hook(code: int, data: bytes) -> None:
return None
serde = JsonPlusSerializer(
allowed_msgpack_modules=None, __unpack_ext_hook__=custom_hook
)
result = serde.with_msgpack_allowlist([MyDataclass])
assert result._unpack_ext_hook is custom_hook
@pytest.mark.skipif(sys.version_info >= (3, 14), reason="pydantic v1 not on 3.14+")
def test_msgpack_pydantic_v1_allowlist(caplog: pytest.LogCaptureFixture) -> None:
"""Pydantic v1 models in allowlist should deserialize without warnings."""
serde = JsonPlusSerializer(
allowed_msgpack_modules=[
("tests.test_jsonplus", "MyPydanticV1"),
("tests.test_jsonplus", "InnerPydanticV1"),
]
)
obj = MyPydanticV1(foo="test", bar=42, inner=InnerPydanticV1(hello="world"))
caplog.clear()
dumped = serde.dumps_typed(obj)
result = serde.loads_typed(dumped)
assert "unregistered type" not in caplog.text.lower()
assert "blocked" not in caplog.text.lower()
assert result == obj
def test_msgpack_dataclass_allowlist(caplog: pytest.LogCaptureFixture) -> None:
"""Dataclasses in allowlist should deserialize without warnings."""
serde = JsonPlusSerializer(
allowed_msgpack_modules=[
("tests.test_jsonplus", "MyDataclass"),
("tests.test_jsonplus", "InnerDataclass"),
]
)
obj = MyDataclass(foo="test", bar=42, inner=InnerDataclass(hello="world"))
caplog.clear()
dumped = serde.dumps_typed(obj)
result = serde.loads_typed(dumped)
assert "unregistered type" not in caplog.text.lower()
assert "blocked" not in caplog.text.lower()
assert result == obj
def test_msgpack_safe_types_value_equality(caplog: pytest.LogCaptureFixture) -> None:
"""Verify safe types are correctly restored with proper values."""
serde = JsonPlusSerializer(allowed_msgpack_modules=None)
test_cases = [
datetime(2024, 1, 15, 12, 30, 45, 123456),
date(2024, 6, 15),
time(14, 30, 0),
uuid.UUID("12345678-1234-5678-1234-567812345678"),
Decimal("123.456789"),
{1, 2, 3, 4, 5},
frozenset(["a", "b", "c"]),
deque([1, 2, 3]),
IPv4Address("10.0.0.1"),
pathlib.Path("/some/test/path"),
re.compile(r"\d+", re.MULTILINE),
]
for obj in test_cases:
caplog.clear()
dumped = serde.dumps_typed(obj)
result = serde.loads_typed(dumped)
assert "blocked" not in caplog.text.lower(), f"Blocked for {type(obj)}"
# For regex patterns, compare pattern and flags
if isinstance(obj, re.Pattern):
assert result.pattern == obj.pattern
assert result.flags == obj.flags
else:
assert result == obj, f"Value mismatch for {type(obj)}: {result} != {obj}"
def test_msgpack_nested_pydantic_serializes_as_dict(
caplog: pytest.LogCaptureFixture,
) -> None:
"""Nested Pydantic models are serialized via model_dump() as dicts.
This means nested models don't go through the ext hook and don't need
to be in the allowlist - only the outer type does.
"""
# Only allow outer type - inner is serialized as dict via model_dump()
serde = JsonPlusSerializer(
allowed_msgpack_modules=[("tests.test_jsonplus", "MyPydantic")]
)
obj = MyPydantic(foo="test", bar=42, inner=InnerPydantic(hello="world"))
caplog.clear()
dumped = serde.dumps_typed(obj)
result = serde.loads_typed(dumped)
# No blocking should occur - inner is serialized as dict, not ext
assert "blocked" not in caplog.text.lower()
assert result == obj
def test_delta_sentinel_serde_round_trip() -> None:
from langgraph.checkpoint.base import DELTA_SENTINEL
from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer
serde = JsonPlusSerializer()
type_tag, blob = serde.dumps_typed(DELTA_SENTINEL)
assert type_tag == "msgpack"
assert blob # non-empty ext envelope
loaded = serde.loads_typed((type_tag, blob))
assert loaded is DELTA_SENTINEL