fix: honor resource auth actions

Co-authored-by: open-swe[bot] <open-swe@users.noreply.github.com>

libs/langgraph/langgraph/pregel/_checkpoint.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import uuid
-from collections.abc import Callable, Iterable, Mapping
+from collections.abc import Callable, Mapping
 from datetime import datetime, timezone
 from typing import Any, cast
 
@@ -14,7 +14,6 @@ from langgraph.checkpoint.base.id import uuid6
 from langgraph.checkpoint.serde.types import _DeltaSnapshot
 
 from langgraph._internal._config import DELTA_MAX_SUPERSTEPS_SINCE_SNAPSHOT
-from langgraph._internal._constants import PUSH
 from langgraph._internal._typing import MISSING
 from langgraph.channels.base import BaseChannel
 from langgraph.channels.delta import DeltaChannel
@@ -71,81 +70,6 @@ def delta_channels_to_snapshot(
     return result
 
 
-def get_updated_channels_from_tasks(
-    run_tasks: Iterable[Any],
-) -> set[str]:
-    """Channel names written by an update_state superstep (excluding PUSH)."""
-    return {c for task in run_tasks for c, _ in task.writes if c != PUSH}
-
-
-def get_delta_channels_from_all_channels(
-    channels: Mapping[str, BaseChannel],
-) -> set[str]:
-    """DeltaChannels to snapshot on the first update_state of a fresh thread."""
-    return {
-        k
-        for k, ch in channels.items()
-        if isinstance(ch, DeltaChannel) and ch.is_available()
-    }
-
-
-def create_metadata_for_update_state_api(
-    channels: Mapping[str, BaseChannel],
-    updated_channels: set[str],
-    *,
-    prev_metadata: Mapping[str, Any] | None,
-) -> dict[str, tuple[int, int]]:
-    """Advance ``counters_since_delta_snapshot`` for update_state on a non-fresh thread.
-
-    Mirrors the per-superstep counter bump in ``_loop._put_checkpoint``.
-    """
-    prev_counters = dict(
-        (prev_metadata or {}).get("counters_since_delta_snapshot") or {}
-    )
-    new_counters: dict[str, tuple[int, int]] = {}
-    for ch_name, ch in channels.items():
-        if not isinstance(ch, DeltaChannel):
-            continue
-        u, s = prev_counters.get(ch_name, (0, 0))
-        s += 1
-        if ch_name in updated_channels:
-            u += 1
-        new_counters[ch_name] = (u, s)
-    return new_counters
-
-
-def create_checkpoint_plan_for_update_state_api(
-    channels: Mapping[str, BaseChannel],
-    updated_channels: set[str],
-    *,
-    step: int,
-    parents: dict[str, Any],
-    saved_metadata: Mapping[str, Any] | None,
-    is_fresh_thread: bool,
-) -> tuple[set[str], dict[str, Any]]:
-    """Return ``(channels_to_snapshot, metadata)`` for an update_state head."""
-    metadata: dict[str, Any] = {
-        "source": "update",
-        "step": step,
-        "parents": parents,
-    }
-    if is_fresh_thread:
-        return get_delta_channels_from_all_channels(channels), metadata
-
-    new_counters = create_metadata_for_update_state_api(
-        channels,
-        updated_channels,
-        prev_metadata=saved_metadata,
-    )
-    channels_to_snapshot = delta_channels_to_snapshot(channels, new_counters)
-    for k in channels_to_snapshot:
-        new_counters[k] = (0, 0)
-    non_zero = {k: v for k, v in new_counters.items() if v != (0, 0)}
-    if non_zero:
-        metadata["counters_since_delta_snapshot"] = non_zero
-    return channels_to_snapshot, metadata
-
-
 def create_checkpoint(
     checkpoint: Checkpoint,
     channels: Mapping[str, BaseChannel] | None,

libs/langgraph/langgraph/pregel/main.py

@@ -108,6 +108,7 @@ from langgraph.callbacks import (
     get_sync_graph_callback_manager_for_config,
 )
 from langgraph.channels.base import BaseChannel
+from langgraph.channels.delta import DeltaChannel
 from langgraph.channels.topic import Topic
 from langgraph.config import get_config
 from langgraph.constants import END
@@ -132,9 +133,7 @@ from langgraph.pregel._checkpoint import (
     channels_from_checkpoint,
     copy_checkpoint,
     create_checkpoint,
-    create_checkpoint_plan_for_update_state_api,
     empty_checkpoint,
-    get_updated_channels_from_tasks,
 )
 from langgraph.pregel._draw import draw_graph
 from langgraph.pregel._io import map_input, read_channels
@@ -1997,14 +1996,13 @@ class Pregel(
                         },
                     ),
                 )
-            updated_channels = get_updated_channels_from_tasks(run_tasks)
-            if saved is not None:
-                for task_id, task in zip(run_task_ids, run_tasks):
-                    channel_writes = [w for w in task.writes if w[0] != PUSH]
-                    if channel_writes:
-                        checkpointer.put_writes(
-                            checkpoint_config, channel_writes, task_id
-                        )
+            # save task writes
+            for task_id, task in zip(run_task_ids, run_tasks):
+                # channel writes are saved to current checkpoint
+                channel_writes = [w for w in task.writes if w[0] != PUSH]
+                if saved and channel_writes:
+                    checkpointer.put_writes(checkpoint_config, channel_writes, task_id)
+            # apply to checkpoint and save
             apply_writes(
                 checkpoint,
                 channels,
@@ -2012,35 +2010,35 @@ class Pregel(
                 checkpointer.get_next_version,
                 self.trigger_to_nodes,
             )
-            channels_to_snapshot, checkpoint_metadata = (
-                create_checkpoint_plan_for_update_state_api(
-                    channels,
-                    updated_channels,
-                    step=step + 1,
-                    parents=saved.metadata.get("parents", {}) if saved else {},
-                    saved_metadata=saved.metadata if saved else None,
-                    is_fresh_thread=saved is None,
-                )
+            # On a fresh thread there is no ancestor to replay DeltaChannel
+            # writes from, so force a self-contained snapshot in the first
+            # checkpoint instead of relying on ancestor write-replay.
+            delta_snapshot = (
+                {
+                    k
+                    for k, ch in channels.items()
+                    if isinstance(ch, DeltaChannel) and ch.is_available()
+                }
+                if saved is None
+                else None
             )
             checkpoint = create_checkpoint(
-                checkpoint,
-                channels,
-                step + 1,
-                updated_channels=updated_channels if channels_to_snapshot else None,
-                get_next_version=checkpointer.get_next_version
-                if channels_to_snapshot
-                else None,
-                channels_to_snapshot=channels_to_snapshot,
+                checkpoint, channels, step + 1, channels_to_snapshot=delta_snapshot
             )
             next_config = checkpointer.put(
                 checkpoint_config,
                 checkpoint,
-                checkpoint_metadata,
+                {
+                    "source": "update",
+                    "step": step + 1,
+                    "parents": saved.metadata.get("parents", {}) if saved else {},
+                },
                 get_new_channel_versions(
                     checkpoint_previous_versions, checkpoint["channel_versions"]
                 ),
             )
             for task_id, task in zip(run_task_ids, run_tasks):
+                # save push writes
                 if push_writes := [w for w in task.writes if w[0] == PUSH]:
                     checkpointer.put_writes(next_config, push_writes, task_id)
 
@@ -2457,14 +2455,15 @@ class Pregel(
                         },
                     ),
                 )
-            updated_channels = get_updated_channels_from_tasks(run_tasks)
-            if saved is not None:
-                for task_id, task in zip(run_task_ids, run_tasks):
-                    channel_writes = [w for w in task.writes if w[0] != PUSH]
-                    if channel_writes:
-                        await checkpointer.aput_writes(
-                            checkpoint_config, channel_writes, task_id
-                        )
+            # save task writes
+            for task_id, task in zip(run_task_ids, run_tasks):
+                # channel writes are saved to current checkpoint
+                channel_writes = [w for w in task.writes if w[0] != PUSH]
+                if saved and channel_writes:
+                    await checkpointer.aput_writes(
+                        checkpoint_config, channel_writes, task_id
+                    )
+            # apply to checkpoint and save
             apply_writes(
                 checkpoint,
                 channels,
@@ -2472,35 +2471,36 @@ class Pregel(
                 checkpointer.get_next_version,
                 self.trigger_to_nodes,
             )
-            channels_to_snapshot, checkpoint_metadata = (
-                create_checkpoint_plan_for_update_state_api(
-                    channels,
-                    updated_channels,
-                    step=step + 1,
-                    parents=saved.metadata.get("parents", {}) if saved else {},
-                    saved_metadata=saved.metadata if saved else None,
-                    is_fresh_thread=saved is None,
-                )
+            # On a fresh thread there is no ancestor to replay DeltaChannel
+            # writes from, so force a self-contained snapshot in the first
+            # checkpoint instead of relying on ancestor write-replay.
+            delta_snapshot = (
+                {
+                    k
+                    for k, ch in channels.items()
+                    if isinstance(ch, DeltaChannel) and ch.is_available()
+                }
+                if saved is None
+                else None
             )
             checkpoint = create_checkpoint(
-                checkpoint,
-                channels,
-                step + 1,
-                updated_channels=updated_channels if channels_to_snapshot else None,
-                get_next_version=checkpointer.get_next_version
-                if channels_to_snapshot
-                else None,
-                channels_to_snapshot=channels_to_snapshot,
+                checkpoint, channels, step + 1, channels_to_snapshot=delta_snapshot
             )
+            # save checkpoint, after applying writes
             next_config = await checkpointer.aput(
                 checkpoint_config,
                 checkpoint,
-                checkpoint_metadata,
+                {
+                    "source": "update",
+                    "step": step + 1,
+                    "parents": saved.metadata.get("parents", {}) if saved else {},
+                },
                 get_new_channel_versions(
                     checkpoint_previous_versions, checkpoint["channel_versions"]
                 ),
             )
             for task_id, task in zip(run_task_ids, run_tasks):
+                # save push writes
                 if push_writes := [w for w in task.writes if w[0] == PUSH]:
                     await checkpointer.aput_writes(next_config, push_writes, task_id)
             return patch_checkpoint_map(next_config, saved.metadata if saved else None)

libs/langgraph/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "langgraph"
-version = "1.2.9"
+version = "1.2.8"
 description = "Building stateful, multi-actor applications with LLMs"
 authors = []
 requires-python = ">=3.10"

libs/langgraph/tests/test_delta_channel_update_state.py

@@ -1,19 +1,23 @@
 """Tests for `update_state` / `aupdate_state` against `DeltaChannel`.
 
-Regression suite for deepagents#3774 and Postgres read-path compatibility.
+Originally a regression suite for deepagents#3774 — `update_state` on a *fresh*
+thread silently dropped the first write to a `DeltaChannel`-backed channel
+because channel writes were only persisted when a previous checkpoint existed
+and no snapshot was written either, so the checkpoint reconstructed to empty.
 
-Fresh-thread ``update_state`` force-snapshots DeltaChannels (1.2.8). Non-fresh
-``update_state`` persists ``checkpoint_writes`` on the parent, advances
-``counters_since_delta_snapshot`` on the new head, and snapshots when a
-channel reaches ``snapshot_frequency`` (mirroring normal run cadence).
+Fixed by forcing a self-contained `_DeltaSnapshot` blob into the first
+checkpoint on a fresh thread (`saved is None`), so the value is stored inline
+and no ancestor write-replay is required. This keeps the read/replay path
+untouched.
 
 Coverage:
 
-* fresh-thread regression: single ``update_state`` writes a message and reads back
-* non-fresh thread: ``update_state`` after ``invoke``, after another ``update_state``,
-  and ``bulk_update_state`` with multiple per-superstep updates
-* update-by-id end-to-end via ``update_state`` (DeltaChannel reducer semantics)
-* fresh-thread head is snapshotted; non-fresh heads carry delta replay counters
+* fresh-thread regression: single `update_state` writes a message and reads back
+* non-fresh thread: `update_state` after `invoke`, after another `update_state`,
+  and `bulk_update_state` with multiple per-superstep updates
+* update-by-id end-to-end via `update_state` (DeltaChannel reducer semantics)
+* state-history chain shape on a fresh thread (single self-contained update
+  checkpoint with the snapshot inline and no parent)
 """
 
 from typing import Annotated, Any
@@ -21,7 +25,6 @@ from typing import Annotated, Any
 import pytest
 from langchain_core.messages import HumanMessage
 from langgraph.checkpoint.memory import InMemorySaver
-from langgraph.checkpoint.serde.types import _DeltaSnapshot
 from typing_extensions import TypedDict
 
 from langgraph.channels.delta import DeltaChannel
@@ -31,20 +34,13 @@ from langgraph.graph.message import _messages_delta_reducer
 pytestmark = pytest.mark.anyio
 
 
-def _build_graph(
-    checkpointer: InMemorySaver,
-    *,
-    two_nodes: bool = False,
-    snapshot_frequency: int = 1000,
-) -> Any:
+def _build_graph(checkpointer: InMemorySaver, *, two_nodes: bool = False) -> Any:
     """Compile a minimal DeltaChannel-backed `messages` graph.
 
     `two_nodes=True` adds a second writer node so `bulk_update_state` can route
     distinct updates to different `as_node` values within a single superstep.
     """
-    channel = DeltaChannel(
-        _messages_delta_reducer, snapshot_frequency=snapshot_frequency
-    )
+    channel = DeltaChannel(_messages_delta_reducer)
     State = TypedDict("State", {"messages": Annotated[list, channel]})  # type: ignore[call-overload]  # noqa: UP013
 
     def model(state: dict) -> dict:
@@ -94,30 +90,14 @@ async def test_aupdate_state_fresh_thread_delta_channel() -> None:
     assert [m.content for m in state.values["messages"]] == ["hello"]
 
 
-def test_fresh_update_state_head_snapshots_delta_channel() -> None:
-    saver = InMemorySaver()
-    graph = _build_graph(saver)
-    config = {"configurable": {"thread_id": "fresh-head-snapshot"}}
-
-    graph.update_state(
-        config,
-        {"messages": [HumanMessage(content="hello", id="m1")]},
-        as_node="model",
-    )
-
-    head = saver.get_tuple(config)
-    assert head is not None
-    assert isinstance(head.checkpoint["channel_values"].get("messages"), _DeltaSnapshot)
-    assert head.metadata is not None
-    assert "counters_since_delta_snapshot" not in head.metadata
-
-
 # ---------------------------------------------------------------------------
 # Non-fresh thread: update_state after invoke
 # ---------------------------------------------------------------------------
 
 
 def test_update_state_after_invoke_delta_channel() -> None:
+    """The non-fresh-thread path was already working before the fix; pin it
+    down so the forced-snapshot change for fresh threads doesn't regress it."""
     saver = InMemorySaver()
     graph = _build_graph(saver)
     config = {"configurable": {"thread_id": "after-invoke-sync"}}
@@ -133,12 +113,6 @@ def test_update_state_after_invoke_delta_channel() -> None:
     assert [m.content for m in state.values["messages"]] == ["seed", "appended"]
     assert [m.id for m in state.values["messages"]] == ["m1", "m2"]
 
-    head = saver.get_tuple(config)
-    assert head is not None
-    assert "messages" not in head.checkpoint["channel_values"]
-    assert head.metadata is not None
-    assert head.metadata["counters_since_delta_snapshot"]["messages"] == [2, 4]
-
 
 async def test_aupdate_state_after_invoke_delta_channel() -> None:
     saver = InMemorySaver()
@@ -162,6 +136,9 @@ async def test_aupdate_state_after_invoke_delta_channel() -> None:
 
 
 def test_consecutive_update_states_delta_channel() -> None:
+    """First update_state forces a self-contained snapshot seed; the second
+    sees a real parent (`saved is not None`) and anchors its writes under that
+    seed. Both messages must round-trip in chronological order."""
     saver = InMemorySaver()
     graph = _build_graph(saver)
     config = {"configurable": {"thread_id": "consecutive-sync"}}
@@ -181,39 +158,6 @@ def test_consecutive_update_states_delta_channel() -> None:
     assert [m.content for m in state.values["messages"]] == ["first", "second"]
     assert [m.id for m in state.values["messages"]] == ["m1", "m2"]
 
-    head = saver.get_tuple(config)
-    assert head is not None
-    assert "messages" not in head.checkpoint["channel_values"]
-    assert head.metadata is not None
-    assert head.metadata["counters_since_delta_snapshot"]["messages"] == [1, 1]
-
-
-def test_update_state_snapshots_at_frequency() -> None:
-    """Non-fresh update_state snapshots when counters reach snapshot_frequency."""
-    saver = InMemorySaver()
-    graph = _build_graph(saver, snapshot_frequency=1)
-    config = {"configurable": {"thread_id": "snapshot-at-freq"}}
-
-    graph.update_state(
-        config,
-        {"messages": [HumanMessage(content="first", id="m1")]},
-        as_node="model",
-    )
-    graph.update_state(
-        config,
-        {"messages": [HumanMessage(content="second", id="m2")]},
-        as_node="model",
-    )
-
-    state = graph.get_state(config)
-    assert [m.content for m in state.values["messages"]] == ["first", "second"]
-
-    head = saver.get_tuple(config)
-    assert head is not None
-    assert isinstance(head.checkpoint["channel_values"].get("messages"), _DeltaSnapshot)
-    assert head.metadata is not None
-    assert "counters_since_delta_snapshot" not in head.metadata
-
 
 async def test_aconsecutive_update_states_delta_channel() -> None:
     saver = InMemorySaver()
@@ -311,7 +255,7 @@ def test_bulk_update_state_multi_task_per_superstep_delta_channel() -> None:
 
 
 # ---------------------------------------------------------------------------
-# Public-API observation of fresh-thread checkpoint shape
+# Public-API observation of the forced-snapshot mechanism
 # ---------------------------------------------------------------------------
 
 

libs/langgraph/uv.lock

@@ -1438,7 +1438,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.9"
+version = "1.2.8"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },

libs/prebuilt/uv.lock

@@ -285,7 +285,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.9"
+version = "1.2.8"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },

libs/sdk-py/langgraph_sdk/auth/__init__.py

@@ -392,7 +392,7 @@ class _ResourceOn(typing.Generic[VCreate, VRead, VUpdate, VDelete, VSearch]):
     def __call__(
         self,
         *,
-        resources: str | Sequence[str],
+        resources: str | Sequence[str] | None = None,
         actions: str | Sequence[str] | None = None,
     ) -> Callable[
         [_ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]],
@@ -416,25 +416,38 @@ class _ResourceOn(typing.Generic[VCreate, VRead, VUpdate, VDelete, VSearch]):
             _ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch],
         ]
     ):
-        if fn is not None:
-            _validate_handler(fn)
-            return typing.cast(
-                "_ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]",
-                _register_handler(self.auth, self.resource, "*", fn),
-            )
-
-        def decorator(
+        def register(
             handler: _ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch],
         ) -> _ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]:
             _validate_handler(handler)
-            return typing.cast(
-                "_ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]",
-                _register_handler(self.auth, self.resource, "*", handler),
-            )
+            if isinstance(resources, str):
+                resource_list = [resources]
+            else:
+                resource_list = (
+                    list(resources) if resources is not None else [self.resource]
+                )
+            if resource_list != [self.resource]:
+                raise ValueError(
+                    f"Resource-specific decorator for {self.resource!r} cannot "
+                    f"register handlers for {resource_list!r}. Use @auth.on(...) "
+                    "for multiple resources."
+                )
+            if isinstance(actions, str):
+                action_list = [actions]
+            else:
+                action_list = list(actions) if actions is not None else ["*"]
+            for action in action_list:
+                _register_handler(self.auth, self.resource, action, handler)
+            return handler
 
-        # Accept keyword-only parameters for future filtering behavior; referenced to satisfy linters.
-        _ = resources, actions
-        return decorator
+        if fn is not None:
+            return register(
+                typing.cast(
+                    "_ActionHandler[VCreate | VUpdate | VRead | VDelete | VSearch]",
+                    fn,
+                )
+            )
+        return register
 
 
 class _AssistantsOn(

libs/sdk-py/tests/test_auth.py

@@ -0,0 +1,44 @@
+import pytest
+
+from langgraph_sdk import Auth
+
+
+def _handler():
+    async def handler(ctx, value):
+        return ctx is not None and value is not None
+
+    return handler
+
+
+def test_resource_decorator_registers_specific_actions():
+    auth = Auth()
+
+    handler = auth.on.threads(actions=["read", "search"])(_handler())
+
+    assert auth._handlers == {
+        ("threads", "read"): [handler],
+        ("threads", "search"): [handler],
+    }
+
+
+def test_resource_decorator_registers_single_action():
+    auth = Auth()
+
+    handler = auth.on.threads(actions="read")(_handler())
+
+    assert auth._handlers == {("threads", "read"): [handler]}
+
+
+def test_resource_decorator_without_actions_registers_resource_wildcard():
+    auth = Auth()
+
+    handler = auth.on.threads(_handler())
+
+    assert auth._handlers == {("threads", "*"): [handler]}
+
+
+def test_resource_decorator_rejects_mismatched_resources():
+    auth = Auth()
+
+    with pytest.raises(ValueError, match=r"Use @auth\.on"):
+        auth.on.threads(resources="assistants", actions="read")(_handler())

libs/sdk-py/uv.lock

@@ -298,7 +298,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.2.9"
+version = "1.2.8"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },