Merge branch 'main' into fix/add-assistant-id-to-auth-types

AGENTS.md

@@ -53,5 +53,3 @@ sdk-js (standalone)
 ```
 
 Changes to a library may impact all of its dependents shown above.
-
-- Do NOT use Sphinx-style double backtick formatting (` ``code`` `). Use single backticks (`` `code` ``) for inline code references in docstrings and comments.

CLAUDE.md

@@ -53,5 +53,3 @@ sdk-js (standalone)
 ```
 
 Changes to a library may impact all of its dependents shown above.
-
-- Do NOT use Sphinx-style double backtick formatting (` ``code`` `). Use single backticks (`` `code` ``) for inline code references in docstrings and comments.

libs/langgraph/langgraph/pregel/_messages.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
 from collections.abc import AsyncIterator, Callable, Iterator, Sequence
-from dataclasses import fields, is_dataclass
 from typing import (
     Any,
     TypeVar,
@@ -12,7 +11,6 @@ from uuid import UUID, uuid4
 from langchain_core.callbacks import BaseCallbackHandler
 from langchain_core.messages import BaseMessage
 from langchain_core.outputs import ChatGeneration, ChatGenerationChunk, LLMResult
-from pydantic import BaseModel
 
 from langgraph._internal._constants import NS_SEP
 from langgraph.constants import TAG_HIDDEN, TAG_NOSTREAM
@@ -28,17 +26,6 @@ T = TypeVar("T")
 Meta = tuple[tuple[str, ...], dict[str, Any]]
 
 
-def _state_values(obj: Any) -> Sequence[Any]:
-    """Extract top-level field values from a state object (dict, BaseModel, or dataclass)."""
-    if isinstance(obj, dict):
-        return list(obj.values())
-    elif isinstance(obj, BaseModel):
-        return [getattr(obj, k) for k in type(obj).model_fields]
-    elif is_dataclass(obj) and not isinstance(obj, type):
-        return [getattr(obj, f.name) for f in fields(obj)]
-    return ()
-
-
 class StreamMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
     """A callback handler that implements stream_mode=messages.
 
@@ -103,14 +90,26 @@ class StreamMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
             for value in response:
                 if isinstance(value, BaseMessage):
                     self._emit(meta, value, dedupe=True)
-        else:
-            for value in _state_values(response):
+        elif isinstance(response, dict):
+            for value in response.values():
                 if isinstance(value, BaseMessage):
                     self._emit(meta, value, dedupe=True)
                 elif isinstance(value, Sequence):
                     for item in value:
                         if isinstance(item, BaseMessage):
                             self._emit(meta, item, dedupe=True)
+        elif hasattr(response, "__dir__") and callable(response.__dir__):
+            for key in dir(response):
+                try:
+                    value = getattr(response, key)
+                    if isinstance(value, BaseMessage):
+                        self._emit(meta, value, dedupe=True)
+                    elif isinstance(value, Sequence):
+                        for item in value:
+                            if isinstance(item, BaseMessage):
+                                self._emit(meta, item, dedupe=True)
+                except AttributeError:
+                    pass
 
     def tap_output_aiter(
         self, run_id: UUID, output: AsyncIterator[T]
@@ -204,15 +203,16 @@ class StreamMessagesHandler(BaseCallbackHandler, _StreamingCallbackHandler):
             if not self.subgraphs and len(ns) > 0:
                 return
             self.metadata[run_id] = (ns, metadata)
-            for value in _state_values(inputs):
-                if isinstance(value, BaseMessage):
-                    if value.id is not None:
-                        self.seen.add(value.id)
-                elif isinstance(value, Sequence) and not isinstance(value, str):
-                    for item in value:
-                        if isinstance(item, BaseMessage):
-                            if item.id is not None:
-                                self.seen.add(item.id)
+            if isinstance(inputs, dict):
+                for key, value in inputs.items():
+                    if isinstance(value, BaseMessage):
+                        if value.id is not None:
+                            self.seen.add(value.id)
+                    elif isinstance(value, Sequence) and not isinstance(value, str):
+                        for item in value:
+                            if isinstance(item, BaseMessage):
+                                if item.id is not None:
+                                    self.seen.add(item.id)
 
     def on_chain_end(
         self,

libs/langgraph/langgraph/pregel/_runner.py

@@ -565,7 +565,7 @@ def _call(
         if fut := next(
             (
                 f
-                for f, t in list(futures().items())  # type: ignore[union-attr]
+                for f, t in futures().items()  # type: ignore[union-attr]
                 if t is not None and t == next_task.id
             ),
             None,
@@ -708,7 +708,7 @@ async def _acall_impl(
             if fut := next(
                 (
                     f
-                    for f, t in list(futures().items())  # type: ignore[union-attr]
+                    for f, t in futures().items()  # type: ignore[union-attr]
                     if t is not None and t == next_task.id
                 ),
                 None,

libs/langgraph/pyproject.toml

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

libs/langgraph/tests/test_pregel.py

@@ -16,7 +16,7 @@ from typing import Annotated, Any, Literal, get_type_hints
 
 import pytest
 from langchain_core.language_models import GenericFakeChatModel
-from langchain_core.messages import AIMessage, AnyMessage, HumanMessage
+from langchain_core.messages import AnyMessage
 from langchain_core.runnables import (
     RunnableConfig,
     RunnableLambda,
@@ -1264,20 +1264,18 @@ def test_imp_task(
     }
 
     thread1 = {"configurable": {"thread_id": "1"}}
-    result = [*graph.stream([0, 1], thread1, durability=durability)]
-    # mapper tasks run concurrently so output order is non-deterministic
-    assert sorted(result[:-1], key=lambda d: str(d)) == [
+    assert [*graph.stream([0, 1], thread1, durability=durability)] == [
         {"mapper": "00"},
         {"mapper": "11"},
+        {
+            "__interrupt__": (
+                Interrupt(
+                    value="question",
+                    id=AnyStr(),
+                ),
+            )
+        },
     ]
-    assert result[-1] == {
-        "__interrupt__": (
-            Interrupt(
-                value="question",
-                id=AnyStr(),
-            ),
-        )
-    }
     assert mapper_calls == 2
 
     assert graph.invoke(Command(resume="answer"), thread1, durability=durability) == [
@@ -6980,93 +6978,6 @@ def test_stream_messages_dedupe_state(sync_checkpointer: BaseCheckpointSaver) ->
     assert chunks[0][1]["langgraph_node"] == "call_model"
 
 
-def test_stream_messages_dedupe_pydantic_subgraph_interrupt(
-    sync_checkpointer: BaseCheckpointSaver,
-) -> None:
-    """Pydantic BaseModel state should not cause duplicate messages when
-    streaming from subgraphs that use interrupts. Regression test for a bug
-    where ``on_chain_start`` only populated the ``seen`` set for dict inputs,
-    skipping Pydantic model inputs entirely."""
-
-    class PydanticState(BaseModel):
-        messages: Annotated[list[AnyMessage], add_messages] = Field(
-            default_factory=list
-        )
-
-    def subgraph_proposal(state) -> Command[Literal["subgraph_approval"]]:
-        return Command(
-            goto="subgraph_approval",
-            update={"messages": [AIMessage(content="Proposal", id="proposal_msg")]},
-        )
-
-    def subgraph_approval(state) -> Command[Literal["__end__"]]:
-        resume_value = interrupt({"message": "Waiting for approval"})
-        user_msg = resume_value.get("user_message", "")
-        msgs = [HumanMessage(content=user_msg)] if user_msg else []
-        return Command(goto="__end__", update={"messages": msgs})
-
-    subgraph = (
-        StateGraph(PydanticState)
-        .add_node("proposal", subgraph_proposal)
-        .add_node("subgraph_approval", subgraph_approval)
-        .add_edge(START, "proposal")
-        .compile(checkpointer=sync_checkpointer)
-    )
-
-    def finalize(state) -> Command[Literal["__end__"]]:
-        return Command(
-            goto="__end__",
-            update={"messages": [AIMessage(content="Finalized", id="finalize_msg")]},
-        )
-
-    graph = (
-        StateGraph(PydanticState)
-        .add_node("subgraph", subgraph)
-        .add_node("finalize", finalize)
-        .add_edge(START, "subgraph")
-        .add_edge("subgraph", "finalize")
-        .compile(checkpointer=sync_checkpointer)
-    )
-
-    thread1 = {"configurable": {"thread_id": "1"}}
-
-    # First stream: should hit interrupt after proposal
-    chunks_req0 = [
-        (ns, chunk)
-        for ns, chunk in graph.stream(
-            {"messages": [HumanMessage(content="Create a proposal")]},
-            thread1,
-            stream_mode="messages",
-            subgraphs=True,
-        )
-    ]
-
-    msg_ids_req0 = {chunk[0].id for _, chunk in chunks_req0}
-    assert "proposal_msg" in msg_ids_req0
-
-    # Verify interrupted
-    state = graph.get_state(thread1)
-    assert state.next
-
-    # Second stream: resume — should NOT duplicate messages from first stream
-    chunks_req1 = [
-        (ns, chunk)
-        for ns, chunk in graph.stream(
-            Command(resume={"user_message": "Yes"}),
-            thread1,
-            stream_mode="messages",
-            subgraphs=True,
-        )
-    ]
-
-    msg_ids_req1 = {chunk[0].id for _, chunk in chunks_req1}
-    assert "finalize_msg" in msg_ids_req1
-
-    # The key assertion: no message IDs from request 0 should appear in request 1
-    duplicates = msg_ids_req0 & msg_ids_req1
-    assert not duplicates, f"Duplicate message IDs across requests: {duplicates}"
-
-
 def test_interrupt_subgraph_reenter_checkpointer_true(
     sync_checkpointer: BaseCheckpointSaver,
 ) -> None:

libs/langgraph/tests/test_pregel_async.py

@@ -2262,20 +2262,18 @@ async def test_imp_task(
 
     tracer = FakeTracer()
     thread1 = {"configurable": {"thread_id": "1"}, "callbacks": [tracer]}
-    result = [c async for c in graph.astream([0, 1], thread1, durability=durability)]
-    # mapper tasks run concurrently so output order is non-deterministic
-    assert sorted(result[:-1], key=lambda d: str(d)) == [
+    assert [c async for c in graph.astream([0, 1], thread1, durability=durability)] == [
         {"mapper": "00"},
         {"mapper": "11"},
+        {
+            "__interrupt__": (
+                Interrupt(
+                    value="question",
+                    id=AnyStr(),
+                ),
+            )
+        },
     ]
-    assert result[-1] == {
-        "__interrupt__": (
-            Interrupt(
-                value="question",
-                id=AnyStr(),
-            ),
-        )
-    }
     assert mapper_calls == 2
     assert len(tracer.runs) == 1
     assert len(tracer.runs[0].child_runs) == 1

libs/langgraph/uv.lock

@@ -1372,7 +1372,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.0.8"
+version = "1.0.7"
 source = { editable = "." }
 dependencies = [
     { name = "langchain-core" },

libs/prebuilt/uv.lock

@@ -268,7 +268,7 @@ wheels = [
 
 [[package]]
 name = "langgraph"
-version = "1.0.8"
+version = "1.0.7"
 source = { editable = "../langgraph" }
 dependencies = [
     { name = "langchain-core" },

libs/sdk-py/langgraph_sdk/__init__.py

@@ -3,6 +3,6 @@ from langgraph_sdk.client import get_client, get_sync_client
 from langgraph_sdk.encryption import Encryption
 from langgraph_sdk.encryption.types import EncryptionContext
 
-__version__ = "0.3.4"
+__version__ = "0.3.3"
 
 __all__ = ["Auth", "Encryption", "EncryptionContext", "get_client", "get_sync_client"]

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

@@ -737,16 +737,21 @@ class CronsCreate(typing.TypedDict, total=False):
 
         ```python
         create_params = {
+            "assistant_id": UUID("123e4567-e89b-12d3-a456-426614173999")
             "payload": {"key": "value"},
             "schedule": "0 0 * * *",
             "cron_id": UUID("123e4567-e89b-12d3-a456-426614174000"),
             "thread_id": UUID("123e4567-e89b-12d3-a456-426614174001"),
             "user_id": "user123",
-            "end_time": datetime(2024, 3, 16, 10, 0, 0)
+            "end_time": datetime(2024, 3, 16, 10, 0, 0),
+            "enabled": true,
         }
         ```
     """
 
+    assistant_id: UUID
+    """Unique identifier for the assistant."""
+
     payload: dict[str, typing.Any]
     """Payload for the cron job."""
 
@@ -765,6 +770,9 @@ class CronsCreate(typing.TypedDict, total=False):
     end_time: datetime | None
     """typing.Optional end time for the cron job."""
 
+    enabled: bool | None
+    """typing.Optional enabled status of the cron job."""
+
 
 class CronsDelete(typing.TypedDict):
     """Payload for deleting a cron job.
@@ -807,7 +815,8 @@ class CronsUpdate(typing.TypedDict, total=False):
         update_params = {
             "cron_id": UUID("123e4567-e89b-12d3-a456-426614174000"),
             "payload": {"key": "value"},
-            "schedule": "0 0 * * *"
+            "schedule": "0 0 * * *",
+            "enabled": true
         }
         ```
     """
@@ -821,6 +830,9 @@ class CronsUpdate(typing.TypedDict, total=False):
     schedule: str | None
     """typing.Optional schedule to update."""
 
+    enabled: bool | None
+    """typing.Optional enabled status of the cron job."""
+
 
 class CronsSearch(typing.TypedDict, total=False):
     """Payload for searching cron jobs.