fix: honor resource auth actions

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

libs/sdk-py/langgraph_sdk/_async/stream.py

@@ -911,34 +911,6 @@ class _SubgraphsProjection:
     def __aiter__(self) -> AsyncIterator[ScopedStreamHandle]:
         return self._subgraphs_iter()
 
-    @staticmethod
-    def _put_root_message(root_inbox: asyncio.Queue[Event | None], item: Event) -> None:
-        if root_inbox.maxsize > 0 and root_inbox.qsize() >= root_inbox.maxsize - 1:
-            raise RuntimeError(
-                "Root messages inbox exceeded max_queue_size while buffering "
-                "root-scope messages. Iterate thread.messages concurrently "
-                "or increase max_queue_size."
-            )
-        try:
-            root_inbox.put_nowait(item)
-        except asyncio.QueueFull as exc:
-            raise RuntimeError(
-                "Root messages inbox exceeded max_queue_size while buffering "
-                "root-scope messages. Iterate thread.messages concurrently "
-                "or increase max_queue_size."
-            ) from exc
-
-    @staticmethod
-    def _signal_root_inbox_closed(root_inbox: asyncio.Queue[Event | None]) -> None:
-        try:
-            root_inbox.put_nowait(None)
-        except asyncio.QueueFull as exc:
-            raise RuntimeError(
-                "Root messages inbox exceeded max_queue_size while closing "
-                "root-scope messages. Iterate thread.messages concurrently "
-                "or increase max_queue_size."
-            ) from exc
-
     async def _subgraphs_iter(self) -> AsyncGenerator[ScopedStreamHandle, None]:
         if self._thread._transport is None:
             raise RuntimeError("AsyncThreadStream not entered - use `async with`.")
@@ -971,7 +943,7 @@ class _SubgraphsProjection:
                     and item.get("method") == "messages"
                     and tuple(_event_namespace(params_field)) == self._scope
                 ):
-                    self._put_root_message(root_inbox, item)
+                    root_inbox.put_nowait(item)
                 for handle in decoder.feed(item):
                     yield handle
         finally:
@@ -989,7 +961,7 @@ class _SubgraphsProjection:
                     handle._finish(terminal_status)
             self._thread._unregister_subscription(sub.id)
             if root_inbox is not None:
-                self._signal_root_inbox_closed(root_inbox)
+                root_inbox.put_nowait(None)
 
 
 class ToolCallHandle:
@@ -1363,8 +1335,7 @@ class AsyncThreadStream:
         that arrive at namespace `[]` before `thread.messages` has subscribed.
         """
         if self._root_messages_inbox is None:
-            maxsize = self._max_queue_size + 1 if self._max_queue_size > 0 else 0
-            self._root_messages_inbox = asyncio.Queue(maxsize=maxsize)
+            self._root_messages_inbox = asyncio.Queue()
         return self._root_messages_inbox
 
     def _register_active_message_stream(self, stream: AsyncChatModelStream) -> None:

libs/sdk-py/langgraph_sdk/_sync/stream.py

@@ -954,34 +954,6 @@ class _SyncSubgraphsProjection:
     def __iter__(self) -> Iterator[SyncScopedStreamHandle]:
         return self._subgraphs_iter()
 
-    @staticmethod
-    def _put_root_message(root_inbox: queue.Queue[Event | None], item: Event) -> None:
-        if root_inbox.maxsize > 0 and root_inbox.qsize() >= root_inbox.maxsize - 1:
-            raise RuntimeError(
-                "Root messages inbox exceeded max_queue_size while buffering "
-                "root-scope messages. Iterate thread.messages concurrently "
-                "or increase max_queue_size."
-            )
-        try:
-            root_inbox.put_nowait(item)
-        except queue.Full as exc:
-            raise RuntimeError(
-                "Root messages inbox exceeded max_queue_size while buffering "
-                "root-scope messages. Iterate thread.messages concurrently "
-                "or increase max_queue_size."
-            ) from exc
-
-    @staticmethod
-    def _signal_root_inbox_closed(root_inbox: queue.Queue[Event | None]) -> None:
-        try:
-            root_inbox.put_nowait(None)
-        except queue.Full as exc:
-            raise RuntimeError(
-                "Root messages inbox exceeded max_queue_size while closing "
-                "root-scope messages. Iterate thread.messages concurrently "
-                "or increase max_queue_size."
-            ) from exc
-
     def _subgraphs_iter(self) -> Iterator[SyncScopedStreamHandle]:
         if self._thread._transport is None:
             raise RuntimeError("SyncThreadStream not entered — use `with`.")
@@ -1014,7 +986,7 @@ class _SyncSubgraphsProjection:
                     and item.get("method") == "messages"
                     and tuple(_event_namespace(params_field)) == self._scope
                 ):
-                    self._put_root_message(root_inbox, item)
+                    root_inbox.put_nowait(item)
                 for handle in decoder.feed(cast(dict[str, Any], item)):
                     yield handle
         finally:
@@ -1035,7 +1007,7 @@ class _SyncSubgraphsProjection:
                     handle._finish(terminal_status)
             self._thread._unregister_subscription(sub.id)
             if root_inbox is not None:
-                self._signal_root_inbox_closed(root_inbox)
+                root_inbox.put_nowait(None)
 
 
 class _SyncExtensionsProjection:
@@ -1250,7 +1222,7 @@ class SyncThreadStream:
 
     def _activate_root_messages_inbox(self) -> queue.Queue[Event | None]:
         if self._root_messages_inbox is None:
-            self._root_messages_inbox = queue.Queue(maxsize=1025)
+            self._root_messages_inbox = queue.Queue()
         return self._root_messages_inbox
 
     def _register_active_message_stream(self, stream: ChatModelStream) -> None:

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/streaming/test_scoped_handles.py

@@ -2,13 +2,7 @@
 
 from __future__ import annotations
 
-import asyncio
-from typing import cast
-from unittest.mock import MagicMock
-
 import httpx
-import pytest
-from langchain_protocol import Event
 
 from langgraph_sdk._async.http import HttpClient
 from langgraph_sdk._async.threads import ThreadsClient
@@ -453,56 +447,6 @@ def test_scoped_handle_inboxes_bounded_by_max_queue_size():
     assert handle._tasks_inbox.maxsize == 16
 
 
-def test_root_messages_inbox_bounded_by_max_queue_size():
-    """Root messages inbox must reserve one terminal sentinel slot."""
-    from langgraph_sdk._async.stream import AsyncThreadStream
-
-    thread = AsyncThreadStream(
-        http=MagicMock(),
-        thread_id="t-1",
-        assistant_id="agent",
-        max_queue_size=16,
-    )
-
-    inbox = thread._activate_root_messages_inbox()
-
-    assert inbox.maxsize == 17
-
-
-def test_subgraphs_root_message_overflow_raises_runtime_error():
-    """Overflowing the root messages inbox must fail explicitly."""
-    from langgraph_sdk._async.stream import _SubgraphsProjection
-
-    inbox: asyncio.Queue[Event | None] = asyncio.Queue(maxsize=2)
-    inbox.put_nowait(cast(Event, message_start_event(seq=1, message_id="msg-1")))
-
-    with pytest.raises(RuntimeError, match="Root messages inbox exceeded"):
-        _SubgraphsProjection._put_root_message(
-            inbox,
-            cast(
-                Event,
-                message_text_delta_event(seq=2, text="overflow", message_id="msg-1"),
-            ),
-        )
-
-
-def test_subgraphs_root_message_close_preserves_full_inbox():
-    """Closing a full root messages inbox must not evict buffered events."""
-    from langgraph_sdk._async.stream import _SubgraphsProjection
-
-    inbox: asyncio.Queue[Event | None] = asyncio.Queue(maxsize=3)
-    first = cast(Event, message_start_event(seq=1, message_id="msg-1"))
-    second = cast(Event, message_finish_event(seq=2, message_id="msg-1"))
-    inbox.put_nowait(first)
-    inbox.put_nowait(second)
-
-    _SubgraphsProjection._signal_root_inbox_closed(inbox)
-
-    assert inbox.get_nowait() is first
-    assert inbox.get_nowait() is second
-    assert inbox.get_nowait() is None
-
-
 async def test_child_handle_inherits_max_queue_size_from_parent():
     """Grandchild ScopedStreamHandles created by _HandleSubgraphsProjection
     inherit the parent's max_queue_size so all queues are consistently bounded."""

libs/sdk-py/tests/streaming/test_sync_projections.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import queue
 from typing import Any, cast
 
 import httpx
@@ -399,53 +398,6 @@ def test_sync_tool_calls_run_error_fails_active_handle():
 # ---------------------------------------------------------------------------
 
 
-def test_sync_root_messages_inbox_is_bounded():
-    """Root messages inbox must reserve one terminal sentinel slot."""
-    fake = SyncFakeServer()
-    fake.script([lifecycle_completed_event(seq=1)])
-    fake.set_state({})
-    with httpx.Client(transport=fake.transport, base_url="http://test") as raw:
-        threads = SyncThreadsClient(SyncHttpClient(raw))
-        with threads.stream(thread_id="t-1", assistant_id="agent") as thread:
-            inbox = thread._activate_root_messages_inbox()
-
-    assert inbox.maxsize == 1025
-
-
-def test_sync_subgraphs_root_message_overflow_raises_runtime_error():
-    """Overflowing the root messages inbox must fail explicitly."""
-    from langgraph_sdk._sync.stream import _SyncSubgraphsProjection
-
-    inbox: queue.Queue[Event | None] = queue.Queue(maxsize=2)
-    inbox.put_nowait(cast(Event, message_start_event(seq=1, message_id="msg-1")))
-
-    with pytest.raises(RuntimeError, match="Root messages inbox exceeded"):
-        _SyncSubgraphsProjection._put_root_message(
-            inbox,
-            cast(
-                Event,
-                message_text_delta_event(seq=2, text="overflow", message_id="msg-1"),
-            ),
-        )
-
-
-def test_sync_subgraphs_root_message_close_preserves_full_inbox():
-    """Closing a full root messages inbox must not evict buffered events."""
-    from langgraph_sdk._sync.stream import _SyncSubgraphsProjection
-
-    inbox: queue.Queue[Event | None] = queue.Queue(maxsize=3)
-    first = cast(Event, message_start_event(seq=1, message_id="msg-1"))
-    second = cast(Event, message_finish_event(seq=2, message_id="msg-1"))
-    inbox.put_nowait(first)
-    inbox.put_nowait(second)
-
-    _SyncSubgraphsProjection._signal_root_inbox_closed(inbox)
-
-    assert inbox.get_nowait() is first
-    assert inbox.get_nowait() is second
-    assert inbox.get_nowait() is None
-
-
 def test_sync_drain_messages_inbox_pre_dispatches_before_yield():
     """When draining the root inbox, str(message.text) must work immediately on yield."""
     fake = SyncFakeServer()

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())