Script 'mail_helper' called by obssrc
Hello community,
here is the log from the commit of package python-langgraph-sdk for
openSUSE:Factory checked in at 2026-10-31 23:10:43
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Comparing /work/SRC/openSUSE:Factory/python-langgraph-sdk (Old)
and /work/SRC/openSUSE:Factory/.python-langgraph-sdk.new.1255 (New)
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Package is "python-langgraph-sdk"
Sat Oct 31 23:10:43 2026 rev:11 rq:1394126 version:0.4.11
Changes:
--------
---
/work/SRC/openSUSE:Factory/python-langgraph-sdk/python-langgraph-sdk.changes
2026-10-27 16:58:32.471551411 +0100
+++
/work/SRC/openSUSE:Factory/.python-langgraph-sdk.new.1255/python-langgraph-sdk.changes
2026-10-31 23:13:38.935075157 +0100
@@ -1,0 +2,13 @@
+Sat Oct 31 13:50:12 UTC 2026 - Martin Pluskal <[email protected]>
+
+- Update to 0.4.11:
+ * Retry failed stream reconnects with backoff; retry
+ runs.wait, join and cancel(wait=True) reconnects
+ * Preserve SSE data and cursor state across chunks; reset
+ the reconnect cursor on every empty id
+ * Warn when a stream reconnects without a Last-Event-ID
+ * Match the Auth.on overload to optional resources
+ * No new runtime dependencies
+ * Spec-cleaner modernization: %files %python_files
+
+-------------------------------------------------------------------
Old:
----
langgraph_sdk-0.4.10.tar.gz
New:
----
langgraph_sdk-0.4.11.tar.gz
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Other differences:
------------------
++++++ python-langgraph-sdk.spec ++++++
--- /var/tmp/diff_new_pack.MgMZcg/_old 2026-10-31 23:13:40.224129429 +0100
+++ /var/tmp/diff_new_pack.MgMZcg/_new 2026-10-31 23:13:40.226129513 +0100
@@ -18,7 +18,7 @@
%{?sle15_python_module_pythons}
Name: python-langgraph-sdk
-Version: 0.4.10
+Version: 0.4.11
Release: 0
Summary: SDK for interacting with the LangGraph API
License: MIT
@@ -73,7 +73,7 @@
# still imported at collection time, so skip the directory outright.
%pytest tests --ignore=tests/integration
-%files %{python_files}
+%files %python_files
%doc README.md
%license LICENSE
%{python_sitelib}/langgraph_sdk
++++++ langgraph_sdk-0.4.10.tar.gz -> langgraph_sdk-0.4.11.tar.gz ++++++
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/langgraph_sdk-0.4.10/PKG-INFO
new/langgraph_sdk-0.4.11/PKG-INFO
--- old/langgraph_sdk-0.4.10/PKG-INFO 2020-02-02 01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.11/PKG-INFO 2020-02-02 01:00:00.000000000 +0100
@@ -1,6 +1,6 @@
Metadata-Version: 2.5
Name: langgraph-sdk
-Version: 0.4.10
+Version: 0.4.11
Summary: SDK for interacting with LangGraph API
Project-URL: Source,
https://github.com/langchain-ai/langgraph/tree/main/libs/sdk-py
Project-URL: Twitter, https://x.com/langchain_oss
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/langgraph_sdk-0.4.10/langgraph_sdk/__init__.py
new/langgraph_sdk-0.4.11/langgraph_sdk/__init__.py
--- old/langgraph_sdk-0.4.10/langgraph_sdk/__init__.py 2020-02-02
01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.11/langgraph_sdk/__init__.py 2020-02-02
01:00:00.000000000 +0100
@@ -5,7 +5,7 @@
from langgraph_sdk.priority import Priority
from langgraph_sdk.priority.types import PriorityContext, PrioritySettings
-__version__ = "0.4.10"
+__version__ = "0.4.11"
__all__ = [
"Auth",
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/langgraph_sdk-0.4.10/langgraph_sdk/_async/http.py
new/langgraph_sdk-0.4.11/langgraph_sdk/_async/http.py
--- old/langgraph_sdk-0.4.10/langgraph_sdk/_async/http.py 2020-02-02
01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.11/langgraph_sdk/_async/http.py 2020-02-02
01:00:00.000000000 +0100
@@ -14,9 +14,14 @@
from langgraph_sdk._shared.utilities import (
_orjson_default,
+ _reconnect_delay,
_validate_reconnect_location,
)
-from langgraph_sdk.errors import _araise_for_status_typed
+from langgraph_sdk.errors import (
+ InternalServerError,
+ RateLimitError,
+ _araise_for_status_typed,
+)
from langgraph_sdk.schema import QueryParamTypes, StreamPart
from langgraph_sdk.sse import SSEDecoder, aiter_lines_raw
@@ -146,7 +151,8 @@
on_response: Callable[[httpx.Response], None] | None = None,
reconnect_limit: int = 5,
) -> Any:
- """Send a request that automatically reconnects to Location header."""
+ """Send a request that reconnects to its Location header if the
response is
+ cut off, with backoff between attempts."""
request_headers, content = await _aencode_json(json)
if headers:
request_headers.update(headers)
@@ -176,13 +182,27 @@
stacklevel=2,
)
await r.aclose()
- return await self.request_reconnect(
- loc,
- "GET",
- headers=headers,
- # don't pass on_response so it's only called once
- reconnect_limit=reconnect_limit - 1,
- )
+ for attempt in range(reconnect_limit):
+ await asyncio.sleep(_reconnect_delay(attempt, base=1.0,
cap=5.0))
+ # A restart or a deploy looks like a refused connection, a
+ # dropped body, a 429 or a 5xx; other errors won't go away.
+ try:
+ return await self.request_reconnect(
+ loc,
+ "GET",
+ headers=headers,
+ # don't pass on_response so it's only called once
+ reconnect_limit=0,
+ )
+ except httpx.HTTPStatusError as err:
+ status = err.response.status_code
+ if attempt == reconnect_limit - 1 or (
+ status != 429 and status < 500
+ ):
+ raise
+ except httpx.TransportError:
+ if attempt == reconnect_limit - 1:
+ raise
async def stream(
self,
@@ -217,72 +237,104 @@
current_headers = dict(
request_headers if reconnect_path is None else
reconnect_headers
)
+ if last_event_id is not None:
+ current_headers = {
+ key: value
+ for key, value in current_headers.items()
+ if key.lower() != "last-event-id"
+ }
if last_event_id:
current_headers["Last-Event-ID"] = last_event_id
+ if reconnect_path is not None and not any(
+ key.lower() == "last-event-id" and value
+ for key, value in current_headers.items()
+ ):
+ warnings.warn(
+ "Reconnecting without a Last-Event-ID, so events sent
while "
+ "disconnected are not replayed. Only runs created with "
+ "stream_resumable=True send the event IDs a reconnect
resumes "
+ "from.",
+ stacklevel=2,
+ )
current_method = method if reconnect_path is None else "GET"
current_content = content if reconnect_path is None else None
current_params = params if reconnect_path is None else None
- retry = False
- async with self.client.stream(
- current_method,
- reconnect_path or path,
- headers=current_headers,
- content=current_content,
- params=current_params,
- ) as res:
- if reconnect_path is None and on_response:
- on_response(res)
- # check status
- await _araise_for_status_typed(res)
- # check content type
- content_type = res.headers.get("content-type",
"").partition(";")[0]
- if "text/event-stream" not in content_type:
- raise httpx.TransportError(
- "Expected response header Content-Type to contain
'text/event-stream', "
- f"got {content_type!r}"
- )
-
- reconnect_location = res.headers.get("location")
- if reconnect_location:
- _validate_reconnect_location(
- self.client.base_url, reconnect_location
- )
- reconnect_path = reconnect_location
-
- # parse SSE
- decoder = SSEDecoder()
- try:
- async for line in aiter_lines_raw(res):
- sse = decoder.decode(line=cast("bytes",
line).rstrip(b"\n"))
- if sse is not None:
+ reconnect_cause: httpx.HTTPError | None = None
+ try:
+ async with self.client.stream(
+ current_method,
+ reconnect_path or path,
+ headers=current_headers,
+ content=current_content,
+ params=current_params,
+ ) as res:
+ if reconnect_path is None and on_response:
+ on_response(res)
+ # check status
+ await _araise_for_status_typed(res)
+ # check content type
+ content_type = res.headers.get("content-type",
"").partition(";")[0]
+ if "text/event-stream" not in content_type:
+ raise httpx.TransportError(
+ "Expected response header Content-Type to contain
'text/event-stream', "
+ f"got {content_type!r}"
+ )
+
+ reconnect_location = res.headers.get("location")
+ if reconnect_location:
+ _validate_reconnect_location(
+ self.client.base_url, reconnect_location
+ )
+ reconnect_path = reconnect_location
+
+ # parse SSE
+ decoder = SSEDecoder(last_event_id=last_event_id)
+ try:
+ async for line in aiter_lines_raw(res):
+ sse = decoder.decode(line=cast("bytes",
line).rstrip(b"\n"))
+ if sse is not None:
+ if decoder.last_event_id is not None:
+ last_event_id = decoder.last_event_id
+ if sse.event or decoder.has_data:
+ # Progress since the last drop earns a
fresh
+ # set of reconnect attempts.
+ reconnect_attempts = 0
+ yield sse
+ except httpx.HTTPError as err:
+ # httpx.TransportError inherits from HTTPError, so
transient
+ # disconnects during streaming land here.
+ if reconnect_path is None:
+ raise
+ reconnect_cause = err
+ else:
+ if sse := decoder.decode(b""):
if decoder.last_event_id is not None:
last_event_id = decoder.last_event_id
- if sse.event or sse.data is not None:
+ if sse.event or decoder.has_data:
+ # decoder.decode(b"") flushes the in-flight
event and may
+ # return an empty placeholder when there is no
pending
+ # message. Skip these no-op events so the
stream doesn't
+ # emit a trailing blank item after reconnects.
yield sse
- except httpx.HTTPError:
- # httpx.TransportError inherits from HTTPError, so
transient
- # disconnects during streaming land here.
- if reconnect_path is None:
- raise
- retry = True
- else:
- if sse := decoder.decode(b""):
- if decoder.last_event_id is not None:
- last_event_id = decoder.last_event_id
- if sse.event or sse.data is not None:
- # decoder.decode(b"") flushes the in-flight event
and may
- # return an empty placeholder when there is no
pending
- # message. Skip these no-op events so the stream
doesn't
- # emit a trailing blank item after reconnects.
- yield sse
- if retry:
+ except (httpx.TransportError, RateLimitError, InternalServerError)
as err:
+ # A reconnect that can't open (connection refused, 429, 5xx or
a
+ # non-SSE body) usually means the server is restarting or
+ # mid-deploy, so retry it like a drop. Before the server sends
a
+ # Location there is nothing to reconnect to, so raise.
+ if reconnect_path is None:
+ raise
+ reconnect_cause = err
+ if reconnect_cause is not None:
reconnect_attempts += 1
if reconnect_attempts > max_reconnect_attempts:
raise httpx.TransportError(
"Exceeded maximum SSE reconnection attempts"
- )
+ ) from reconnect_cause
+ await asyncio.sleep(
+ _reconnect_delay(reconnect_attempts - 1, base=1.0, cap=5.0)
+ )
continue
break
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/langgraph_sdk-0.4.10/langgraph_sdk/_async/stream.py
new/langgraph_sdk-0.4.11/langgraph_sdk/_async/stream.py
--- old/langgraph_sdk-0.4.10/langgraph_sdk/_async/stream.py 2020-02-02
01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.11/langgraph_sdk/_async/stream.py 2020-02-02
01:00:00.000000000 +0100
@@ -16,7 +16,6 @@
import asyncio
import contextlib
import logging
-import random
from collections.abc import AsyncGenerator, AsyncIterator, Generator, Mapping
from dataclasses import dataclass, field
from typing import Any, Literal, TypedDict, cast
@@ -25,7 +24,7 @@
from langchain_protocol import Event, SubscribeParams
from langgraph_sdk._async.http import HttpClient
-from langgraph_sdk._shared.utilities import _quote_path_param
+from langgraph_sdk._shared.utilities import _quote_path_param, _reconnect_delay
from langgraph_sdk.schema import LangSmithTracing, QueryParamTypes
from langgraph_sdk.stream.controller import _SeenEventIds
from langgraph_sdk.stream.decoders import (
@@ -1691,11 +1690,13 @@
async def _reconnect_sleep(self, attempt: int) -> None:
"""Sleep with exponential backoff and jitter for reconnect attempt
`attempt`."""
- base = self._shared_reconnect_backoff_base
- cap = self._shared_reconnect_backoff_cap
- delay = min(cap, base * (2**attempt))
- jitter = random.uniform(0, delay * 0.25)
- await asyncio.sleep(delay + jitter)
+ await asyncio.sleep(
+ _reconnect_delay(
+ attempt,
+ base=self._shared_reconnect_backoff_base,
+ cap=self._shared_reconnect_backoff_cap,
+ )
+ )
async def _reconnect_shared_stream(self) -> bool:
"""Attempt to reopen the shared stream after a post-ready transport
drop.
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore'
old/langgraph_sdk-0.4.10/langgraph_sdk/_shared/utilities.py
new/langgraph_sdk-0.4.11/langgraph_sdk/_shared/utilities.py
--- old/langgraph_sdk-0.4.10/langgraph_sdk/_shared/utilities.py 2020-02-02
01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.11/langgraph_sdk/_shared/utilities.py 2020-02-02
01:00:00.000000000 +0100
@@ -4,6 +4,7 @@
import functools
import os
+import random
import re
from collections.abc import Mapping
from datetime import tzinfo
@@ -194,6 +195,15 @@
return location
+def _reconnect_delay(attempt: int, *, base: float, cap: float) -> float:
+ """Seconds to wait before reconnect attempt `attempt` (0-based).
+
+ Exponential backoff capped at `cap`, plus up to 25% jitter.
+ """
+ delay = min(cap, base * (2**attempt))
+ return delay + random.uniform(0, delay * 0.25)
+
+
def _provided_vals(d: Mapping[str, Any]) -> dict[str, Any]:
return {k: v for k, v in d.items() if v is not None}
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/langgraph_sdk-0.4.10/langgraph_sdk/_sync/http.py
new/langgraph_sdk-0.4.11/langgraph_sdk/_sync/http.py
--- old/langgraph_sdk-0.4.10/langgraph_sdk/_sync/http.py 2020-02-02
01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.11/langgraph_sdk/_sync/http.py 2020-02-02
01:00:00.000000000 +0100
@@ -4,6 +4,7 @@
import logging
import sys
+import time
import warnings
from collections.abc import Callable, Iterator, Mapping
from typing import Any, cast
@@ -13,9 +14,14 @@
from langgraph_sdk._shared.utilities import (
_orjson_default,
+ _reconnect_delay,
_validate_reconnect_location,
)
-from langgraph_sdk.errors import _raise_for_status_typed
+from langgraph_sdk.errors import (
+ InternalServerError,
+ RateLimitError,
+ _raise_for_status_typed,
+)
from langgraph_sdk.schema import QueryParamTypes, StreamPart
from langgraph_sdk.sse import SSEDecoder, iter_lines_raw
@@ -146,7 +152,8 @@
on_response: Callable[[httpx.Response], None] | None = None,
reconnect_limit: int = 5,
) -> Any:
- """Send a request that automatically reconnects to Location header."""
+ """Send a request that reconnects to its Location header if the
response is
+ cut off, with backoff between attempts."""
request_headers, content = _encode_json(json)
if headers:
request_headers.update(headers)
@@ -176,13 +183,27 @@
stacklevel=2,
)
r.close()
- return self.request_reconnect(
- loc,
- "GET",
- headers=headers,
- # don't pass on_response so it's only called once
- reconnect_limit=reconnect_limit - 1,
- )
+ for attempt in range(reconnect_limit):
+ time.sleep(_reconnect_delay(attempt, base=1.0, cap=5.0))
+ # A restart or a deploy looks like a refused connection, a
+ # dropped body, a 429 or a 5xx; other errors won't go away.
+ try:
+ return self.request_reconnect(
+ loc,
+ "GET",
+ headers=headers,
+ # don't pass on_response so it's only called once
+ reconnect_limit=0,
+ )
+ except httpx.HTTPStatusError as err:
+ status = err.response.status_code
+ if attempt == reconnect_limit - 1 or (
+ status != 429 and status < 500
+ ):
+ raise
+ except httpx.TransportError:
+ if attempt == reconnect_limit - 1:
+ raise
def stream(
self,
@@ -219,69 +240,99 @@
current_headers = dict(
request_headers if reconnect_path is None else
reconnect_headers
)
+ if last_event_id is not None:
+ current_headers = {
+ key: value
+ for key, value in current_headers.items()
+ if key.lower() != "last-event-id"
+ }
if last_event_id:
current_headers["Last-Event-ID"] = last_event_id
+ if reconnect_path is not None and not any(
+ key.lower() == "last-event-id" and value
+ for key, value in current_headers.items()
+ ):
+ warnings.warn(
+ "Reconnecting without a Last-Event-ID, so events sent
while "
+ "disconnected are not replayed. Only runs created with "
+ "stream_resumable=True send the event IDs a reconnect
resumes "
+ "from.",
+ stacklevel=2,
+ )
current_method = method if reconnect_path is None else "GET"
current_content = content if reconnect_path is None else None
current_params = params if reconnect_path is None else None
- retry = False
- with self.client.stream(
- current_method,
- reconnect_path or path,
- headers=current_headers,
- content=current_content,
- params=current_params,
- ) as res:
- if reconnect_path is None and on_response:
- on_response(res)
- # check status
- _raise_for_status_typed(res)
- # check content type
- content_type = res.headers.get("content-type",
"").partition(";")[0]
- if "text/event-stream" not in content_type:
- raise httpx.TransportError(
- "Expected response header Content-Type to contain
'text/event-stream', "
- f"got {content_type!r}"
- )
-
- reconnect_location = res.headers.get("location")
- if reconnect_location:
- _validate_reconnect_location(
- self.client.base_url, reconnect_location
- )
- reconnect_path = reconnect_location
-
- decoder = SSEDecoder()
- try:
- for line in iter_lines_raw(res):
- sse = decoder.decode(cast(bytes, line).rstrip(b"\n"))
- if sse is not None:
+ reconnect_cause: httpx.HTTPError | None = None
+ try:
+ with self.client.stream(
+ current_method,
+ reconnect_path or path,
+ headers=current_headers,
+ content=current_content,
+ params=current_params,
+ ) as res:
+ if reconnect_path is None and on_response:
+ on_response(res)
+ # check status
+ _raise_for_status_typed(res)
+ # check content type
+ content_type = res.headers.get("content-type",
"").partition(";")[0]
+ if "text/event-stream" not in content_type:
+ raise httpx.TransportError(
+ "Expected response header Content-Type to contain
'text/event-stream', "
+ f"got {content_type!r}"
+ )
+
+ reconnect_location = res.headers.get("location")
+ if reconnect_location:
+ _validate_reconnect_location(
+ self.client.base_url, reconnect_location
+ )
+ reconnect_path = reconnect_location
+
+ decoder = SSEDecoder(last_event_id=last_event_id)
+ try:
+ for line in iter_lines_raw(res):
+ sse = decoder.decode(cast(bytes,
line).rstrip(b"\n"))
+ if sse is not None:
+ if decoder.last_event_id is not None:
+ last_event_id = decoder.last_event_id
+ if sse.event or decoder.has_data:
+ # Progress since the last drop earns a
fresh
+ # set of reconnect attempts.
+ reconnect_attempts = 0
+ yield sse
+ except httpx.HTTPError as err:
+ # httpx.TransportError inherits from HTTPError, so
transient
+ # disconnects during streaming land here.
+ if reconnect_path is None:
+ raise
+ reconnect_cause = err
+ else:
+ if sse := decoder.decode(b""):
if decoder.last_event_id is not None:
last_event_id = decoder.last_event_id
- if sse.event or sse.data is not None:
+ if sse.event or decoder.has_data:
+ # See async stream implementation for
rationale on
+ # skipping empty flush events.
yield sse
- except httpx.HTTPError:
- # httpx.TransportError inherits from HTTPError, so
transient
- # disconnects during streaming land here.
- if reconnect_path is None:
- raise
- retry = True
- else:
- if sse := decoder.decode(b""):
- if decoder.last_event_id is not None:
- last_event_id = decoder.last_event_id
- if sse.event or sse.data is not None:
- # See async stream implementation for rationale on
- # skipping empty flush events.
- yield sse
- if retry:
+ except (httpx.TransportError, RateLimitError, InternalServerError)
as err:
+ # A reconnect that can't open (connection refused, 429, 5xx or
a
+ # non-SSE body) usually means the server is restarting or
+ # mid-deploy, so retry it like a drop. Before the server sends
a
+ # Location there is nothing to reconnect to, so raise.
+ if reconnect_path is None:
+ raise
+ reconnect_cause = err
+ if reconnect_cause is not None:
reconnect_attempts += 1
if reconnect_attempts > max_reconnect_attempts:
raise httpx.TransportError(
"Exceeded maximum SSE reconnection attempts"
- )
+ ) from reconnect_cause
+ time.sleep(_reconnect_delay(reconnect_attempts - 1, base=1.0,
cap=5.0))
continue
break
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/langgraph_sdk-0.4.10/langgraph_sdk/auth/__init__.py
new/langgraph_sdk-0.4.11/langgraph_sdk/auth/__init__.py
--- old/langgraph_sdk-0.4.10/langgraph_sdk/auth/__init__.py 2020-02-02
01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.11/langgraph_sdk/auth/__init__.py 2020-02-02
01:00:00.000000000 +0100
@@ -811,7 +811,7 @@
def __call__(
self,
*,
- resources: str | Sequence[str],
+ resources: str | Sequence[str] | None = None,
actions: str | Sequence[str] | None = None,
) -> Callable[[AHO], AHO]: ...
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/langgraph_sdk-0.4.10/langgraph_sdk/sse.py
new/langgraph_sdk-0.4.11/langgraph_sdk/sse.py
--- old/langgraph_sdk-0.4.10/langgraph_sdk/sse.py 2020-02-02
01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.11/langgraph_sdk/sse.py 2020-02-02
01:00:00.000000000 +0100
@@ -30,18 +30,14 @@
# See https://docs.python.org/3/glossary.html#term-universal-newlines
NEWLINE_CHARS = b"\n\r"
- # We always push a trailing `\r` into the next decode iteration.
- if self.trailing_cr:
- text = b"\r" + text
- self.trailing_cr = False
- if text.endswith(b"\r"):
- self.trailing_cr = True
- text = text[:-1]
+ if not text:
+ return []
+ if self.trailing_cr and text.startswith(b"\n"):
+ text = text[1:]
+ self.trailing_cr = text.endswith(b"\r")
if not text:
- # NOTE: the edge case input of empty text doesn't occur in
practice,
- # because other httpx internals filter out this value
- return [] # pragma: no cover
+ return []
trailing_newline = text[-1] in NEWLINE_CHARS
lines = cast(list[BytesLike], text.splitlines())
@@ -66,7 +62,7 @@
return lines
def flush(self) -> list[BytesLike]:
- if not self.buffer and not self.trailing_cr:
+ if not self.buffer:
return []
lines: list[BytesLike] = [self.buffer]
@@ -76,18 +72,25 @@
class SSEDecoder:
- def __init__(self) -> None:
+ def __init__(self, *, last_event_id: str | None = None) -> None:
self._event = ""
self._data = bytearray()
- self._last_event_id: str | None = None
+ self._last_event_id = last_event_id
self._retry: int | None = None
self._at_stream_start = True
+ self._has_data = False
+ self._id_changed = False
+
+ @property
+ def has_data(self) -> bool:
+ """Whether the last decoded block contained a data field."""
+ return self._has_data
@property
def last_event_id(self) -> str | None:
"""Return the last event identifier that was seen.
- `None` until the stream sends an `id` field, and `""` after an empty
+ `None` until an ID is supplied or received, and `""` after an empty
one, which resets the reconnect cursor.
"""
@@ -101,17 +104,26 @@
line = line.removeprefix(b"\xef\xbb\xbf")
if not line:
+ self._has_data = bool(self._data)
if (
not self._event
and not self._data
- and not self._last_event_id
+ and not self._id_changed
and self._retry is None
):
return None
+ body = self._data[:-1]
+ data = None
+ if body and not body.isspace():
+ try:
+ data = orjson.loads(body)
+ except orjson.JSONDecodeError:
+ data = orjson.loads(body.decode(errors="replace"))
+
sse = StreamPart(
event=self._event,
- data=orjson.loads(self._data) if self._data else None, # ty:
ignore[invalid-argument-type]
+ data=data, # ty: ignore[invalid-argument-type]
id=self._last_event_id or None,
)
@@ -119,6 +131,7 @@
self._event = ""
self._data = bytearray()
self._retry = None
+ self._id_changed = False
return sse
@@ -131,16 +144,16 @@
value = value[1:]
if fieldname == b"event":
- self._event = value.decode()
+ self._event = value.decode(errors="replace")
elif fieldname == b"data":
- if self._data:
- self._data.extend(b"\n")
self._data.extend(value)
+ self._data.extend(b"\n")
elif fieldname == b"id":
if b"\0" in value:
pass
else:
- self._last_event_id = value.decode()
+ self._last_event_id = value.decode(errors="replace")
+ self._id_changed = True
elif fieldname == b"retry":
with contextlib.suppress(TypeError, ValueError):
self._retry = int(value)
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore'
old/langgraph_sdk-0.4.10/langgraph_sdk/stream/controller.py
new/langgraph_sdk-0.4.11/langgraph_sdk/stream/controller.py
--- old/langgraph_sdk-0.4.10/langgraph_sdk/stream/controller.py 2020-02-02
01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.11/langgraph_sdk/stream/controller.py 2020-02-02
01:00:00.000000000 +0100
@@ -15,7 +15,6 @@
import asyncio
import contextlib
import logging
-import random
from collections import OrderedDict
from collections.abc import AsyncGenerator, AsyncIterator, Awaitable, Callable
from dataclasses import dataclass, field
@@ -23,6 +22,7 @@
from langchain_protocol import Event, SubscribeParams
+from langgraph_sdk._shared.utilities import _reconnect_delay
from langgraph_sdk.stream.transport import AsyncProtocolTransport,
EventStreamHandle
# ---------------------------------------------------------------------------
@@ -275,11 +275,13 @@
async def _reconnect_sleep(self, attempt: int) -> None:
"""Sleep with exponential backoff and jitter for reconnect attempt
*attempt*."""
- base = self._reconnect_backoff_base
- cap = self._reconnect_backoff_cap
- delay = min(cap, base * (2**attempt))
- jitter = random.uniform(0, delay * 0.25)
- await asyncio.sleep(delay + jitter)
+ await asyncio.sleep(
+ _reconnect_delay(
+ attempt,
+ base=self._reconnect_backoff_base,
+ cap=self._reconnect_backoff_cap,
+ )
+ )
async def _reconnect_shared_stream(self) -> bool:
"""Attempt to reopen the shared stream after a transport drop.
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore'
old/langgraph_sdk-0.4.10/langgraph_sdk/stream/sync_controller.py
new/langgraph_sdk-0.4.11/langgraph_sdk/stream/sync_controller.py
--- old/langgraph_sdk-0.4.10/langgraph_sdk/stream/sync_controller.py
2020-02-02 01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.11/langgraph_sdk/stream/sync_controller.py
2020-02-02 01:00:00.000000000 +0100
@@ -4,7 +4,6 @@
import contextlib
import logging
-import random
import threading
import time
from dataclasses import dataclass, field
@@ -13,6 +12,7 @@
from langchain_protocol import Event, SubscribeParams
+from langgraph_sdk._shared.utilities import _reconnect_delay
from langgraph_sdk.stream.errors import (
is_non_retryable_stream_error,
stream_error_frame,
@@ -282,11 +282,13 @@
def _reconnect_sleep(self, attempt: int) -> None:
"""Sleep with exponential backoff + jitter before a reconnect
attempt."""
- base = self._reconnect_backoff_base
- cap = self._reconnect_backoff_cap
- delay = min(cap, base * (2**attempt))
- jitter = random.uniform(0, delay * 0.25)
- time.sleep(delay + jitter)
+ time.sleep(
+ _reconnect_delay(
+ attempt,
+ base=self._reconnect_backoff_base,
+ cap=self._reconnect_backoff_cap,
+ )
+ )
def _reconnect_shared_stream(self) -> bool:
"""Attempt to reopen the shared stream after a transport drop.
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/langgraph_sdk-0.4.10/tests/conftest.py
new/langgraph_sdk-0.4.11/tests/conftest.py
--- old/langgraph_sdk-0.4.10/tests/conftest.py 1970-01-01 01:00:00.000000000
+0100
+++ new/langgraph_sdk-0.4.11/tests/conftest.py 2020-02-02 01:00:00.000000000
+0100
@@ -0,0 +1,17 @@
+import asyncio
+import time
+
+import pytest
+
+
[email protected]
+def no_sleep(monkeypatch):
+ """Record backoff sleeps without actually waiting."""
+ sleeps: list[float] = []
+
+ async def fake_async_sleep(d):
+ sleeps.append(d)
+
+ monkeypatch.setattr(asyncio, "sleep", fake_async_sleep)
+ monkeypatch.setattr(time, "sleep", lambda d: sleeps.append(d))
+ return sleeps
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/langgraph_sdk-0.4.10/tests/test_auth.py
new/langgraph_sdk-0.4.11/tests/test_auth.py
--- old/langgraph_sdk-0.4.10/tests/test_auth.py 2020-02-02 01:00:00.000000000
+0100
+++ new/langgraph_sdk-0.4.11/tests/test_auth.py 2020-02-02 01:00:00.000000000
+0100
@@ -1,8 +1,49 @@
+import ast
+import inspect
+
import pytest
from langgraph_sdk import Auth
[email protected](
+ "kwargs",
+ [{}, {"actions": "read"}, {"resources": None}],
+)
+def test_global_handler_factory_accepts_optional_resources(kwargs) -> None:
+ auth = Auth()
+ tree = ast.parse(inspect.getsource(type(auth.on)))
+ cls = tree.body[0]
+ assert isinstance(cls, ast.ClassDef)
+ factory_overload = next(
+ node
+ for node in cls.body
+ if isinstance(node, ast.FunctionDef)
+ and node.name == "__call__"
+ and node.decorator_list
+ and any(arg.arg == "resources" for arg in node.args.kwonlyargs)
+ )
+ resources = next(
+ arg for arg in factory_overload.args.kwonlyargs if arg.arg ==
"resources"
+ )
+ assert resources.annotation is not None
+ assert ast.unparse(resources.annotation) == "str | Sequence[str] | None"
+ default = factory_overload.args.kw_defaults[0]
+ assert isinstance(default, ast.Constant) and default.value is None
+
+ @auth.on(**kwargs)
+ async def handler(ctx, value):
+ del ctx, value
+ return None
+
+ if kwargs.get("actions"):
+ assert auth._handlers == {("*", "read"): [handler]}
+ assert auth._global_handlers == []
+ else:
+ assert auth._global_handlers == [handler]
+ assert auth._handlers == {}
+
+
def test_handler_multiple_resources_and_actions() -> None:
auth = Auth()
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/langgraph_sdk-0.4.10/tests/test_client_stream.py
new/langgraph_sdk-0.4.11/tests/test_client_stream.py
--- old/langgraph_sdk-0.4.10/tests/test_client_stream.py 2020-02-02
01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.11/tests/test_client_stream.py 2020-02-02
01:00:00.000000000 +0100
@@ -1,6 +1,6 @@
from __future__ import annotations
-from collections.abc import Iterator, Sequence
+from collections.abc import Callable, Iterator, Sequence
from pathlib import Path
from typing import Any
@@ -13,6 +13,7 @@
from langgraph_sdk._shared.utilities import _sse_to_v2_dict
from langgraph_sdk._sync.runs import _wrap_stream_v2_sync
from langgraph_sdk.client import HttpClient, SyncHttpClient
+from langgraph_sdk.errors import InternalServerError, NotFoundError
from langgraph_sdk.schema import (
CheckpointPayload,
CheckpointsStreamPart,
@@ -130,6 +131,110 @@
assert decoder.decode(b"") == StreamPart(event="values", data=1, id=None)
# ty: ignore[invalid-argument-type]
+def test_control_blocks_do_not_repeat_after_a_cursor_update():
+ decoder = SSEDecoder()
+ decoder.decode(b"id: cursor")
+ assert decoder.decode(b"") == StreamPart(event="", data=None, id="cursor")
# ty: ignore[invalid-argument-type]
+ decoder.decode(b": heartbeat")
+ assert decoder.decode(b"") is None
+ assert decoder.decode(b"") is None
+ decoder.decode(b"id:")
+ assert decoder.decode(b"") == StreamPart(event="", data=None, id=None) #
ty: ignore[invalid-argument-type]
+ assert decoder.last_event_id == ""
+ assert decoder.decode(b"") is None
+
+
+def test_cr_lines_are_emitted_without_waiting_for_another_chunk():
+ decoder = BytesLineDecoder()
+ assert decoder.decode(b"data: 1\r\r") == [b"data: 1", b""]
+ assert decoder.decode(b"\n") == []
+ assert decoder.decode(b"data: 2\r") == [b"data: 2"]
+ assert decoder.decode(b"") == []
+ assert decoder.decode(b"\n\r") == [b""]
+ assert decoder.flush() == []
+
+
+def test_empty_data_lines_preserve_their_newlines():
+ decoder = SSEDecoder()
+ for line in (b"data", b"data: 1", b"data:"):
+ decoder.decode(line)
+ assert decoder._data == b"\n1\n\n"
+ assert decoder.decode(b"") == StreamPart(event="", data=1, id=None) # ty:
ignore[invalid-argument-type]
+ decoder.decode(b"data:")
+ decoder.decode(b"data:")
+ assert decoder.decode(b"") == StreamPart(event="", data=None, id=None) #
ty: ignore[invalid-argument-type]
+ assert decoder.has_data
+ assert decoder.decode(b"") is None
+ assert not decoder.has_data
+
+
+def test_malformed_utf8_is_replaced_in_fields_and_json_strings():
+ decoder = SSEDecoder()
+ for line in (b"event: \xff", b"id: \xff", b'data: "\xff"'):
+ decoder.decode(line)
+ assert decoder.decode(b"") == StreamPart(event="\ufffd", data="\ufffd",
id="\ufffd") # ty: ignore[invalid-argument-type]
+
+
[email protected]("asynchronous", [False, True])
+async def test_null_and_empty_data_are_not_dropped(asynchronous: bool):
+ def handler(_request: httpx.Request) -> httpx.Response:
+ return httpx.Response(
+ 200,
+ headers={"Content-Type": "text/event-stream"},
+ content=b"data: null\n\ndata:\n\n",
+ )
+
+ transport = httpx.MockTransport(handler)
+ if asynchronous:
+ async with httpx.AsyncClient(
+ transport=transport, base_url="https://example.com";
+ ) as client:
+ parts = [part async for part in
HttpClient(client).stream("/stream", "GET")]
+ else:
+ with httpx.Client(
+ transport=transport, base_url="https://example.com";
+ ) as client:
+ parts = list(SyncHttpClient(client).stream("/stream", "GET"))
+ assert parts == [StreamPart(event="", data=None), StreamPart(event="",
data=None)] # ty: ignore[invalid-argument-type]
+
+
[email protected]("asynchronous", [False, True])
+async def test_cursor_is_inherited_after_reconnect(asynchronous: bool):
+ requests: list[httpx.Request] = []
+
+ def handler(request: httpx.Request) -> httpx.Response:
+ requests.append(request)
+ if len(requests) == 1:
+ stream_cls = AsyncListByteStream if asynchronous else
ListByteStream
+ return httpx.Response(
+ 200,
+ headers={"Content-Type": "text/event-stream", "Location":
"/reconnect"},
+ stream=stream_cls(
+ [b"id: cursor\nevent: values\ndata: 1\n\n"],
+ httpx.RemoteProtocolError("incomplete chunked read"),
+ ),
+ )
+ assert (b"Last-Event-ID", b"cursor") in request.headers.raw
+ return httpx.Response(
+ 200,
+ headers={"Content-Type": "text/event-stream"},
+ content=b"event: values\ndata: 2\n\n",
+ )
+
+ transport = httpx.MockTransport(handler)
+ if asynchronous:
+ async with httpx.AsyncClient(
+ transport=transport, base_url="https://example.com";
+ ) as client:
+ parts = [part async for part in
HttpClient(client).stream("/stream", "GET")]
+ else:
+ with httpx.Client(
+ transport=transport, base_url="https://example.com";
+ ) as client:
+ parts = list(SyncHttpClient(client).stream("/stream", "GET"))
+ assert [part.id for part in parts] == ["cursor", "cursor"]
+
+
# --- HTTP client streaming ---
@@ -176,6 +281,7 @@
assert parts == [StreamPart(event="foo", data={"bar": 1})]
[email protected]("no_sleep")
def test_sync_http_client_stream_recovers_after_disconnect():
reconnect_path = "/reconnect"
first_chunks = [
@@ -241,6 +347,7 @@
@pytest.mark.asyncio
[email protected]("no_sleep")
async def test_http_client_stream_recovers_after_disconnect():
reconnect_path = "/reconnect"
first_chunks = [
@@ -312,18 +419,14 @@
]
-def _drop_after_an_empty_id(stream_cls: type, reconnects: list[httpx.Request]):
+def _drop_after(chunks: list[bytes], stream_cls: type, reconnects:
list[httpx.Request]):
def handler(request: httpx.Request) -> httpx.Response:
if request.method == "POST":
return httpx.Response(
200,
headers={"Content-Type": "text/event-stream", "Location":
"/reconnect"},
stream=stream_cls(
- [
- b"id: 1\nevent: values\ndata: 1\n\n",
- b"id:\nevent: values\ndata: 2\n\n",
- ],
- httpx.RemoteProtocolError("incomplete chunked read"),
+ chunks, httpx.RemoteProtocolError("incomplete chunked
read")
),
)
reconnects.append(request)
@@ -336,26 +439,409 @@
return handler
-def test_sync_stream_reconnects_without_a_cursor_after_an_empty_id():
+def _sync_reconnect(
+ chunks: list[bytes], headers: dict[str, str] | None
+) -> httpx.Request:
reconnects: list[httpx.Request] = []
- transport = httpx.MockTransport(_drop_after_an_empty_id(ListByteStream,
reconnects))
+ transport = httpx.MockTransport(_drop_after(chunks, ListByteStream,
reconnects))
with httpx.Client(transport=transport, base_url="https://example.com";) as
client:
- list(SyncHttpClient(client).stream("/stream", "POST", json={}))
-
- assert "last-event-id" not in reconnects[0].headers
+ list(SyncHttpClient(client).stream("/stream", "POST", json={},
headers=headers))
+ return reconnects[0]
-async def test_async_stream_reconnects_without_a_cursor_after_an_empty_id():
+async def _async_reconnect(
+ chunks: list[bytes], headers: dict[str, str] | None
+) -> httpx.Request:
reconnects: list[httpx.Request] = []
transport = httpx.MockTransport(
- _drop_after_an_empty_id(AsyncListByteStream, reconnects)
+ _drop_after(chunks, AsyncListByteStream, reconnects)
)
async with httpx.AsyncClient(
transport=transport, base_url="https://example.com";
) as client:
- [part async for part in HttpClient(client).stream("/stream", "POST",
json={})]
+ [
+ part
+ async for part in HttpClient(client).stream(
+ "/stream", "POST", json={}, headers=headers
+ )
+ ]
+ return reconnects[0]
+
+
+_EMPTY_IDS = pytest.mark.parametrize(
+ "empty_id",
+ [b"id:\nevent: values\ndata: 2\n\n", b"id:\n\n"],
+ ids=["with-an-event", "alone"],
+)
+_CALLER_CURSORS = pytest.mark.parametrize(
+ "headers",
+ [None, {"Last-Event-ID": "5"}, {"last-event-id": "5"}],
+ ids=["no-caller-cursor", "caller-cursor", "lowercase-caller-cursor"],
+)
+
+
+@_EMPTY_IDS
+@_CALLER_CURSORS
[email protected]("no_sleep")
+def test_sync_stream_reconnects_without_a_cursor_after_an_empty_id(
+ empty_id: bytes, headers: dict[str, str] | None
+):
+ with pytest.warns(UserWarning, match="Last-Event-ID"):
+ reconnect = _sync_reconnect(
+ [b"id: 1\nevent: values\ndata: 1\n\n", empty_id], headers
+ )
+
+ assert "last-event-id" not in reconnect.headers
+
+
+@_EMPTY_IDS
+@_CALLER_CURSORS
[email protected]("no_sleep")
+async def test_async_stream_reconnects_without_a_cursor_after_an_empty_id(
+ empty_id: bytes, headers: dict[str, str] | None
+):
+ with pytest.warns(UserWarning, match="Last-Event-ID"):
+ reconnect = await _async_reconnect(
+ [b"id: 1\nevent: values\ndata: 1\n\n", empty_id], headers
+ )
+
+ assert "last-event-id" not in reconnect.headers
+
+
[email protected]("no_sleep")
+def test_sync_stream_id_replaces_a_lowercase_caller_cursor():
+ reconnect = _sync_reconnect(
+ [b"id: 1\nevent: values\ndata: 1\n\n"], {"last-event-id": "5"}
+ )
+
+ assert reconnect.headers.get_list("last-event-id") == ["1"]
+
+
[email protected]("no_sleep")
+async def test_async_stream_id_replaces_a_lowercase_caller_cursor():
+ reconnect = await _async_reconnect(
+ [b"id: 1\nevent: values\ndata: 1\n\n"], {"last-event-id": "5"}
+ )
+
+ assert reconnect.headers.get_list("last-event-id") == ["1"]
+
+
[email protected]("no_sleep")
+def test_sync_stream_keeps_the_caller_cursor_until_the_stream_sends_an_id():
+ reconnect = _sync_reconnect([b"event: values\ndata: 1\n\n"],
{"Last-Event-ID": "5"})
+
+ assert reconnect.headers.get_list("last-event-id") == ["5"]
+
+
[email protected]("no_sleep")
+async def
test_async_stream_keeps_the_caller_cursor_until_the_stream_sends_an_id():
+ reconnect = await _async_reconnect(
+ [b"event: values\ndata: 1\n\n"], {"Last-Event-ID": "5"}
+ )
+
+ assert reconnect.headers.get_list("last-event-id") == ["5"]
+
+
+# A reconnect that sends no Last-Event-ID can't replay what was sent while it
was
+# disconnected, so it warns.
+_RECONNECT_CURSORS = pytest.mark.parametrize(
+ ("chunks", "headers", "warns"),
+ [
+ ([b"event: values\ndata: 1\n\n"], None, True),
+ ([b"id: 1\nevent: values\ndata: 1\n\n"], None, False),
+ ([b"event: values\ndata: 1\n\n"], {"Last-Event-ID": "5"}, False),
+ ],
+ ids=["no-cursor", "stream-cursor", "caller-cursor"],
+)
+
+
+@_RECONNECT_CURSORS
[email protected]("no_sleep")
+def test_sync_reconnect_warns_only_without_a_cursor(
+ chunks: list[bytes],
+ headers: dict[str, str] | None,
+ warns: bool,
+ recwarn: pytest.WarningsRecorder,
+):
+ _sync_reconnect(chunks, headers)
+
+ assert any("Last-Event-ID" in str(w.message) for w in recwarn) is warns
+
+
+@_RECONNECT_CURSORS
[email protected]("no_sleep")
+async def test_async_reconnect_warns_only_without_a_cursor(
+ chunks: list[bytes],
+ headers: dict[str, str] | None,
+ warns: bool,
+ recwarn: pytest.WarningsRecorder,
+):
+ await _async_reconnect(chunks, headers)
+
+ assert any("Last-Event-ID" in str(w.message) for w in recwarn) is warns
+
+
+# --- reconnects survive a server restart or a deploy ---
+
+SSE = {"Content-Type": "text/event-stream"}
+DROP = httpx.RemoteProtocolError("incomplete chunked read")
+
+
+def _event(i: int) -> bytes:
+ return f"id: {i}\nevent: values\ndata: {i}\n\n".encode()
+
+
+def _scripted(steps: list[Callable[[httpx.Request], httpx.Response]]):
+ """Handler that plays one step per request and records each request."""
+ seen: list[httpx.Request] = []
+
+ def handler(request: httpx.Request) -> httpx.Response:
+ seen.append(request)
+ if len(seen) > len(steps):
+ raise AssertionError("unexpected request")
+ return steps[len(seen) - 1](request)
+
+ return handler, seen
+
+
+def _ok(stream_cls, chunks, *, drop=False, location=None):
+ headers = dict(SSE)
+ if location:
+ headers["Location"] = location
+
+ def step(_request):
+ return httpx.Response(
+ 200, headers=headers, stream=stream_cls(chunks, DROP if drop else
None)
+ )
+
+ return step
+
+
+def _status(code):
+ return lambda request: httpx.Response(
+ code, json={"detail": "upstream"}, request=request
+ )
+
+
+def _connect_error(request):
+ raise httpx.ConnectError("[Errno 61] Connection refused", request=request)
+
+
+def _not_sse(request):
+ return httpx.Response(
+ 200, headers={"Content-Type": "text/html"}, text="<html>",
request=request
+ )
+
+
+def _transient_steps(stream_cls, failure):
+ """POST drops after event 1; first reconnect fails; second succeeds."""
+ return [
+ _ok(stream_cls, [_event(1)], drop=True, location="/reconnect"),
+ failure,
+ _ok(stream_cls, [_event(2), b"event: end\ndata: null\n\n"]),
+ ]
+
+
+def _long_stream_steps(stream_cls, drops):
+ """POST + `drops` reconnects, each delivering one event then dropping."""
+ steps = [_ok(stream_cls, [_event(0)], drop=True, location="/reconnect")]
+ for i in range(1, drops):
+ steps.append(_ok(stream_cls, [_event(i)], drop=True))
+ steps.append(_ok(stream_cls, [_event(drops), b"event: end\ndata:
null\n\n"]))
+ return steps
+
+
+def _run_sync(steps):
+ handler, seen = _scripted(steps)
+ with httpx.Client(
+ transport=httpx.MockTransport(handler), base_url="https://example.com";
+ ) as client:
+ parts = list(SyncHttpClient(client).stream("/stream", "POST", json={}))
+ return parts, seen
+
+
+async def _run_async(steps):
+ sync_handler, seen = _scripted(steps)
+
+ async def handler(request):
+ return sync_handler(request)
+
+ async with httpx.AsyncClient(
+ transport=httpx.MockTransport(handler), base_url="https://example.com";
+ ) as client:
+ parts = [p async for p in HttpClient(client).stream("/stream", "POST",
json={})]
+ return parts, seen
+
+
+FAILURES = {
+ "429": _status(429),
+ "502": _status(502),
+ "503": _status(503),
+ "connect_error": _connect_error,
+ "not_sse": _not_sse,
+}
+
+
+# A reconnect that hits a 429 / 502 / 503 / connect error / non-SSE body, then
+# succeeds, should recover.
+
+
[email protected]("no_sleep")
[email protected]("failure", FAILURES.values(), ids=FAILURES.keys())
+def test_sync_reconnect_recovers_from_transient_failure(failure):
+ parts, seen = _run_sync(_transient_steps(ListByteStream, failure))
+ assert [p.event for p in parts] == ["values", "values", "end"]
+ assert len(seen) == 3
+
+
[email protected]("no_sleep")
[email protected]("failure", FAILURES.values(), ids=FAILURES.keys())
+async def test_async_reconnect_recovers_from_transient_failure(failure):
+ parts, seen = await _run_async(_transient_steps(AsyncListByteStream,
failure))
+ assert [p.event for p in parts] == ["values", "values", "end"]
+ assert len(seen) == 3
+
+
+# Any other 4xx on a reconnect won't go away by retrying, so it raises.
+
+
+def _client_error_steps(stream_cls):
+ return [
+ _ok(stream_cls, [_event(1)], drop=True, location="/reconnect"),
+ _status(404),
+ ]
+
+
[email protected]("no_sleep")
+def test_sync_reconnect_does_not_retry_a_client_error():
+ with pytest.raises(NotFoundError):
+ _run_sync(_client_error_steps(ListByteStream))
+
+
[email protected]("no_sleep")
+async def test_async_reconnect_does_not_retry_a_client_error():
+ with pytest.raises(NotFoundError):
+ await _run_async(_client_error_steps(AsyncListByteStream))
+
+
+# The first request isn't a reconnect, so it raises even when it names a
Location.
+
+
+def _first_request_503(request):
+ return httpx.Response(
+ 503,
+ headers={"Location": "/reconnect"},
+ json={"detail": "upstream"},
+ request=request,
+ )
+
+
[email protected]("no_sleep")
+def test_sync_first_request_failure_is_not_retried():
+ with pytest.raises(InternalServerError):
+ _run_sync([_first_request_503])
+
+
[email protected]("no_sleep")
+async def test_async_first_request_failure_is_not_retried():
+ with pytest.raises(InternalServerError):
+ await _run_async([_first_request_503])
+
+
+# A server that stays down gets five reconnects, backing off 1, 2, 4, 5 and 5s
+# (each plus up to 25% jitter), so about 20s, before the stream gives up. The
+# error it gives up with is caused by the last failure, so callers see why.
+
+GIVE_UP = "Exceeded maximum SSE reconnection attempts"
+
+
+def _server_down_steps(stream_cls):
+ return [
+ _ok(stream_cls, [_event(1)], drop=True, location="/reconnect"),
+ *[_status(502)] * 5,
+ ]
+
+
+def _assert_backed_off_for_about_20s(sleeps: list[float]) -> None:
+ assert len(sleeps) == 5
+ for slept, delay in zip(sleeps, [1, 2, 4, 5, 5], strict=True):
+ assert delay <= slept <= delay * 1.25
+
+
+def test_sync_reconnect_gives_up_after_five_backed_off_attempts(no_sleep):
+ with pytest.raises(httpx.TransportError, match=GIVE_UP) as exc_info:
+ _run_sync(_server_down_steps(ListByteStream))
+ assert isinstance(exc_info.value.__cause__, InternalServerError)
+ _assert_backed_off_for_about_20s(no_sleep)
+
+
+async def
test_async_reconnect_gives_up_after_five_backed_off_attempts(no_sleep):
+ with pytest.raises(httpx.TransportError, match=GIVE_UP) as exc_info:
+ await _run_async(_server_down_steps(AsyncListByteStream))
+ assert isinstance(exc_info.value.__cause__, InternalServerError)
+ _assert_backed_off_for_about_20s(no_sleep)
+
+
+# A long stream that drops 6 times, each reconnect succeeding and making
+# progress, should not run out of attempts.
+
+
[email protected]("no_sleep")
+def test_sync_long_stream_survives_many_spaced_out_drops():
+ parts, _ = _run_sync(_long_stream_steps(ListByteStream, drops=6))
+ assert [p.data for p in parts] == [0, 1, 2, 3, 4, 5, 6, None]
+
+
[email protected]("no_sleep")
+async def test_async_long_stream_survives_many_spaced_out_drops():
+ parts, _ = await _run_async(_long_stream_steps(AsyncListByteStream,
drops=6))
+ assert [p.data for p in parts] == [0, 1, 2, 3, 4, 5, 6, None]
+
+
+# Each reconnect resumes after the newest event the stream got, so a reconnect
+# loop on a resumable run replays only what it missed. A failed reconnect keeps
+# the cursor for the next one.
+RECONNECT_LOOPS = {
+ "progress_between_drops": (
+ lambda stream_cls: _long_stream_steps(stream_cls, drops=3),
+ [None, "0", "1", "2"],
+ ),
+ "failed_reconnect": (
+ lambda stream_cls: _transient_steps(stream_cls, _status(502)),
+ [None, "1", "1"],
+ ),
+}
+
+
[email protected]("no_sleep")
[email protected](
+ ("steps", "cursors"), RECONNECT_LOOPS.values(), ids=RECONNECT_LOOPS.keys()
+)
+def test_sync_reconnect_loop_resumes_after_the_newest_event(steps, cursors):
+ _, seen = _run_sync(steps(ListByteStream))
+ assert [r.headers.get("last-event-id") for r in seen] == cursors
+
+
[email protected]("no_sleep")
[email protected](
+ ("steps", "cursors"), RECONNECT_LOOPS.values(), ids=RECONNECT_LOOPS.keys()
+)
+async def test_async_reconnect_loop_resumes_after_the_newest_event(steps,
cursors):
+ _, seen = await _run_async(steps(AsyncListByteStream))
+ assert [r.headers.get("last-event-id") for r in seen] == cursors
+
+
+# Reconnects should back off rather than hammer the server.
+
+
+def test_sync_reconnect_backs_off(no_sleep):
+ _run_sync(_long_stream_steps(ListByteStream, drops=2))
+ assert len(no_sleep) == 2 and all(d > 0 for d in no_sleep)
+
- assert "last-event-id" not in reconnects[0].headers
+async def test_async_reconnect_backs_off(no_sleep):
+ await _run_async(_long_stream_steps(AsyncListByteStream, drops=2))
+ assert len(no_sleep) == 2 and all(d > 0 for d in no_sleep)
# --- _sse_to_v2_dict conversion ---
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/langgraph_sdk-0.4.10/tests/test_runs_client.py
new/langgraph_sdk-0.4.11/tests/test_runs_client.py
--- old/langgraph_sdk-0.4.10/tests/test_runs_client.py 2020-02-02
01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.11/tests/test_runs_client.py 2020-02-02
01:00:00.000000000 +0100
@@ -1,7 +1,8 @@
from __future__ import annotations
import json
-from collections.abc import AsyncIterator, Iterator
+from collections.abc import AsyncIterator, Callable, Iterator, Sequence
+from typing import Any
import httpx
import pytest
@@ -24,7 +25,16 @@
raise httpx.ReadError("connection dropped")
-def _wait_handler(requests: list[httpx.Request]):
+def _done(_request: httpx.Request) -> httpx.Response:
+ return httpx.Response(200, json={"done": True})
+
+
+def _wait_handler(
+ requests: list[httpx.Request],
+ reconnects: Sequence[Callable[[httpx.Request], httpx.Response]] = (_done,),
+):
+ """POST drops mid-body; the reconnects answer from `reconnects` in
order."""
+
def handler(request: httpx.Request) -> httpx.Response:
requests.append(request)
if request.method == "POST":
@@ -33,7 +43,7 @@
headers={"Location": "/threads/t1/runs/r1/join"},
stream=_BodyDropsMidRead(),
)
- return httpx.Response(200, json={"done": True})
+ return reconnects[len(requests) - 2](request)
return handler
@@ -46,6 +56,7 @@
assert request.read() == b""
[email protected]("no_sleep")
def test_sync_wait_reconnects_with_a_bodyless_get():
requests: list[httpx.Request] = []
transport = httpx.MockTransport(_wait_handler(requests))
@@ -58,6 +69,7 @@
_assert_bodyless_get(requests[1])
[email protected]("no_sleep")
async def test_async_wait_reconnects_with_a_bodyless_get():
requests: list[httpx.Request] = []
transport = httpx.MockTransport(_wait_handler(requests))
@@ -72,6 +84,89 @@
_assert_bodyless_get(requests[1])
+def _connect_error(request: httpx.Request) -> httpx.Response:
+ raise httpx.ConnectError("[Errno 61] Connection refused", request=request)
+
+
+async def _wait(
+ handler: Callable[[httpx.Request], httpx.Response], *, asynchronous: bool
+) -> Any:
+ transport = httpx.MockTransport(handler)
+ if asynchronous:
+ async with httpx.AsyncClient(
+ transport=transport, base_url="https://example.com";
+ ) as client:
+ return await RunsClient(HttpClient(client)).wait("t1", "agent",
input={})
+ with httpx.Client(transport=transport, base_url="https://example.com";) as
client:
+ return SyncRunsClient(SyncHttpClient(client)).wait("t1", "agent",
input={})
+
+
+# What a server restart or a deploy looks like to the reconnect GET: each one
is
+# retried after a backoff.
+RECONNECT_FAILURES = {
+ "429": lambda request: httpx.Response(429, request=request),
+ "503": lambda request: httpx.Response(503, request=request),
+ "connect_error": _connect_error,
+ "body_drops": lambda _request: httpx.Response(
+ 200,
+ headers={"Location": "/threads/t1/runs/r1/join"},
+ stream=_BodyDropsMidRead(),
+ ),
+}
+
+
[email protected]("asynchronous", [False, True])
[email protected](
+ "failure", RECONNECT_FAILURES.values(), ids=RECONNECT_FAILURES.keys()
+)
+async def test_wait_retries_a_failed_reconnect_after_a_backoff(
+ failure, asynchronous, no_sleep
+):
+ requests: list[httpx.Request] = []
+ with pytest.warns(UserWarning, match="attempting reconnect"):
+ result = await _wait(
+ _wait_handler(requests, [failure, _done]),
asynchronous=asynchronous
+ )
+
+ assert result == {"done": True}
+ assert len(requests) == 3
+ assert len(no_sleep) == 2 and min(no_sleep) >= 1, no_sleep
+
+
[email protected]("no_sleep")
[email protected]("asynchronous", [False, True])
+async def test_wait_raises_after_five_failed_reconnects(asynchronous):
+ requests: list[httpx.Request] = []
+ with (
+ pytest.warns(UserWarning, match="attempting reconnect"),
+ pytest.raises(httpx.HTTPStatusError, match="503"),
+ ):
+ await _wait(
+ _wait_handler(requests, [RECONNECT_FAILURES["503"]] * 5),
+ asynchronous=asynchronous,
+ )
+
+ assert len(requests) == 6
+
+
[email protected]("no_sleep")
[email protected]("asynchronous", [False, True])
+async def test_wait_does_not_retry_a_client_error(asynchronous):
+ requests: list[httpx.Request] = []
+ with (
+ pytest.warns(UserWarning, match="attempting reconnect"),
+ pytest.raises(httpx.HTTPStatusError, match="404"),
+ ):
+ await _wait(
+ _wait_handler(
+ requests, [lambda request: httpx.Response(404,
request=request)]
+ ),
+ asynchronous=asynchronous,
+ )
+
+ assert len(requests) == 2
+
+
def test_sync_wait_raises_the_run_error():
requests: list[httpx.Request] = []
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/langgraph_sdk-0.4.10/uv.lock
new/langgraph_sdk-0.4.11/uv.lock
--- old/langgraph_sdk-0.4.10/uv.lock 2020-02-02 01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.11/uv.lock 2020-02-02 01:00:00.000000000 +0100
@@ -301,7 +301,7 @@
[[package]]
name = "langgraph"
-version = "1.2.18"
+version = "1.2.19"
source = { editable = "../langgraph" }
dependencies = [
{ name = "langchain-core" },
