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-17 21:26:59
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Comparing /work/SRC/openSUSE:Factory/python-langgraph-sdk (Old)
and /work/SRC/openSUSE:Factory/.python-langgraph-sdk.new.3698736 (New)
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Package is "python-langgraph-sdk"
Sat Oct 17 21:26:59 2026 rev:8 rq:1385355 version:0.4.8
Changes:
--------
---
/work/SRC/openSUSE:Factory/python-langgraph-sdk/python-langgraph-sdk.changes
2026-10-15 12:25:38.920898128 +0200
+++
/work/SRC/openSUSE:Factory/.python-langgraph-sdk.new.3698736/python-langgraph-sdk.changes
2026-10-17 21:27:03.053006716 +0200
@@ -1,0 +2,9 @@
+Sat Oct 17 06:32:27 UTC 2026 - Martin Pluskal <[email protected]>
+
+- Update to 0.4.8:
+ * Thread-stream protocol error frames surface as errors
+ instead of silent closes
+ * Lifecycle watcher starts with run.start and is guarded
+ against races
+
+-------------------------------------------------------------------
Old:
----
langgraph_sdk-0.4.7.tar.gz
New:
----
langgraph_sdk-0.4.8.tar.gz
++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
Other differences:
------------------
++++++ python-langgraph-sdk.spec ++++++
--- /var/tmp/diff_new_pack.Cm6DIA/_old 2026-10-17 21:27:03.773036758 +0200
+++ /var/tmp/diff_new_pack.Cm6DIA/_new 2026-10-17 21:27:03.777036924 +0200
@@ -18,7 +18,7 @@
%{?sle15_python_module_pythons}
Name: python-langgraph-sdk
-Version: 0.4.7
+Version: 0.4.8
Release: 0
Summary: SDK for interacting with the LangGraph API
License: MIT
++++++ langgraph_sdk-0.4.7.tar.gz -> langgraph_sdk-0.4.8.tar.gz ++++++
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/langgraph_sdk-0.4.7/PKG-INFO
new/langgraph_sdk-0.4.8/PKG-INFO
--- old/langgraph_sdk-0.4.7/PKG-INFO 2020-02-02 01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.8/PKG-INFO 2020-02-02 01:00:00.000000000 +0100
@@ -1,6 +1,6 @@
Metadata-Version: 2.5
Name: langgraph-sdk
-Version: 0.4.7
+Version: 0.4.8
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.7/langgraph_sdk/__init__.py
new/langgraph_sdk-0.4.8/langgraph_sdk/__init__.py
--- old/langgraph_sdk-0.4.7/langgraph_sdk/__init__.py 2020-02-02
01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.8/langgraph_sdk/__init__.py 2020-02-02
01:00:00.000000000 +0100
@@ -3,7 +3,7 @@
from langgraph_sdk.encryption import Encryption
from langgraph_sdk.encryption.types import DecryptResult, EncryptionContext
-__version__ = "0.4.7"
+__version__ = "0.4.8"
__all__ = [
"Auth",
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/langgraph_sdk-0.4.7/langgraph_sdk/_async/stream.py
new/langgraph_sdk-0.4.8/langgraph_sdk/_async/stream.py
--- old/langgraph_sdk-0.4.7/langgraph_sdk/_async/stream.py 2020-02-02
01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.8/langgraph_sdk/_async/stream.py 2020-02-02
01:00:00.000000000 +0100
@@ -15,6 +15,7 @@
import asyncio
import contextlib
+import logging
import random
from collections.abc import AsyncGenerator, AsyncIterator, Generator, Mapping
from dataclasses import dataclass, field
@@ -36,6 +37,11 @@
ToolCallsDecoder,
validate_interleave_channels,
)
+from langgraph_sdk.stream.errors import (
+ is_non_retryable_stream_error,
+ protocol_error,
+ stream_error_frame,
+)
from langgraph_sdk.stream.subscription import compute_union_filter,
infer_channel
from langgraph_sdk.stream.transport import (
AsyncProtocolTransport,
@@ -44,6 +50,8 @@
ProtocolWebSocketTransport,
)
+logger = logging.getLogger(__name__)
+
class InterruptPayload(TypedDict):
"""Payload surfaced when the server requests human input for a thread."""
@@ -193,6 +201,7 @@
if not gate.done():
gate.set_result(None)
self._owner._run_seen = True
+ self._owner._ensure_lifecycle_watcher_running()
return result
except BaseException as err:
# Why: gate MUST reject on any exit type, including CancelledError,
@@ -1203,6 +1212,7 @@
# observes a `completed`/`errored` event.
self._interrupts_lock = asyncio.Lock()
self._lifecycle_watcher_task: asyncio.Task[None] | None = None
+ self._lifecycle_watcher_rejected = False
self._lifecycle_watcher_handle: EventStreamHandle | None = None
self._lifecycle_cursor: int | None = None
self._lifecycle_max_reconnect_attempts = 5
@@ -1261,8 +1271,11 @@
# Create the run-done future here (async context guarantees a running
loop).
self._run_done = asyncio.get_running_loop().create_future()
# Start the lifecycle watcher immediately so reattach and thread.output
- # work without a preceding run.start call.
- self._ensure_lifecycle_watcher_running()
+ # work without a preceding run.start call. A minted thread_id names no
+ # thread until run.start creates it, and with custom auth a stream for
a
+ # thread that doesn't exist yet fails, so run.start starts the watcher.
+ if self._explicit_thread_id:
+ self._ensure_lifecycle_watcher_running()
return self
async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
@@ -1630,6 +1643,13 @@
async for event in self._dedup_iter(shared.events):
if self._closed:
break
+ if (frame_error := stream_error_frame(event)) is not None:
+ logger.warning(
+ "Thread %s event stream failed: %s",
+ self.thread_id,
+ frame_error,
+ )
+ continue
self._observe_event(event)
for sub in list(self._subscriptions.values()):
if matches_subscription(event, sub.params):
@@ -1655,6 +1675,7 @@
if (
err is not None
and not isinstance(err, asyncio.CancelledError)
+ and not is_non_retryable_stream_error(err)
and not self._closed
):
with contextlib.suppress(Exception):
@@ -1681,7 +1702,8 @@
Returns:
`True` if a new stream was opened (caller should resume fanout),
- `False` if all reconnect attempts were exhausted or the controller
+ `False` if all reconnect attempts were exhausted, the server
+ rejected the stream with a non-retryable 4xx, or the controller
was closed in the meantime.
"""
if self._transport is None:
@@ -1702,7 +1724,9 @@
await new_stream.ready
except asyncio.CancelledError:
raise
- except Exception:
+ except Exception as err:
+ if is_non_retryable_stream_error(err):
+ return False
await self._reconnect_sleep(attempt)
continue
self._shared_stream = new_stream
@@ -1800,9 +1824,7 @@
# 202/204 — no body. Caller gets an empty result.
return {}
if response.get("type") == "error":
- code = response.get("error", "unknown")
- message = response.get("message", "")
- raise RuntimeError(f"Protocol error [{code}]: {message}")
+ raise protocol_error(response)
meta = response.get("meta")
if isinstance(meta, dict):
applied_through_seq = meta.get("applied_through_seq")
@@ -1826,8 +1848,15 @@
await asyncio.wait_for(asyncio.shield(gate), timeout=timeout)
def _ensure_lifecycle_watcher_running(self) -> None:
- if self._lifecycle_watcher_task is not None:
+ task = self._lifecycle_watcher_task
+ if task is not None and not self._lifecycle_watcher_rejected:
return
+ if task is not None:
+ # A 4xx stopped the watcher before this run started, typically the
+ # 404 for a thread this run.start just created, so the result it
+ # left isn't this run's.
+ self._run_done = asyncio.get_running_loop().create_future()
+ self._lifecycle_watcher_rejected = False
self._lifecycle_watcher_task = asyncio.create_task(
self._run_lifecycle_watcher()
)
@@ -1850,7 +1879,11 @@
if self._transport is None:
return
reconnect_attempts = 0
+ frame_error: RuntimeError | None = None
while not self._closed:
+ # A stream opened before run.start can 404 for the thread that
+ # run.start creates, so it gets one more try once a run has
started.
+ opened_before_run = not self._run_seen
try:
handle = self._transport.open_event_stream(
self._lifecycle_stream_params()
@@ -1860,6 +1893,7 @@
async for event in handle.events:
if self._closed:
return
+ frame_error = stream_error_frame(event) or frame_error
self._observe_lifecycle_event(event)
await self._apply_lifecycle_event(event)
err = await handle.done
@@ -1868,38 +1902,43 @@
# event. Resolve `_run_done` as errored so awaiters of
# `thread.output` don't hang.
if err is None:
- run_done = self._run_done
- if run_done is not None and not run_done.done():
- run_done.set_result(
- _RunTerminal(
- status="errored",
- error=RuntimeError(
- "lifecycle stream ended before
terminal event"
- ),
- )
+ self._fail_run_done(
+ frame_error
+ or RuntimeError(
+ "lifecycle stream ended before terminal event"
)
+ )
return
reconnect_attempts += 1
- if reconnect_attempts > self._lifecycle_max_reconnect_attempts:
+ if (
+ is_non_retryable_stream_error(err)
+ or reconnect_attempts >
self._lifecycle_max_reconnect_attempts
+ ):
raise err
await asyncio.sleep(0.05)
except asyncio.CancelledError:
raise
except Exception as exc:
+ if is_non_retryable_stream_error(exc):
+ if opened_before_run and self._run_seen:
+ continue
+ self._lifecycle_watcher_rejected = True
+ self._fail_run_done(exc)
+ return
reconnect_attempts += 1
if reconnect_attempts <=
self._lifecycle_max_reconnect_attempts:
await asyncio.sleep(0.05)
continue
- run_done = self._run_done
- if run_done is not None and not run_done.done():
- run_done.set_result(
- _RunTerminal(
- status="errored",
- error=RuntimeError(f"Lifecycle transport failed:
{exc}"),
- )
- )
+ self._fail_run_done(
+ frame_error or RuntimeError(f"Lifecycle transport failed:
{exc}")
+ )
return
+ def _fail_run_done(self, error: BaseException) -> None:
+ run_done = self._run_done
+ if run_done is not None and not run_done.done():
+ run_done.set_result(_RunTerminal(status="errored", error=error))
+
async def _fetch_state(self) -> dict[str, Any]:
"""Fetch the current thread state from the REST endpoint."""
return await self._http.get(
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/langgraph_sdk-0.4.7/langgraph_sdk/_sync/stream.py
new/langgraph_sdk-0.4.8/langgraph_sdk/_sync/stream.py
--- old/langgraph_sdk-0.4.7/langgraph_sdk/_sync/stream.py 2020-02-02
01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.8/langgraph_sdk/_sync/stream.py 2020-02-02
01:00:00.000000000 +0100
@@ -34,6 +34,11 @@
ToolCallsDecoder,
validate_interleave_channels,
)
+from langgraph_sdk.stream.errors import (
+ is_non_retryable_stream_error,
+ protocol_error,
+ stream_error_frame,
+)
from langgraph_sdk.stream.subscription import compute_union_filter,
infer_channel
from langgraph_sdk.stream.sync_controller import (
SyncStreamController,
@@ -237,6 +242,7 @@
controller = self._owner._controller
if controller is not None and controller._run_start_gate is not None:
controller._run_start_gate.set()
+ self._owner._ensure_lifecycle_watcher_running()
return result
def respond(
@@ -1111,6 +1117,7 @@
self.interrupted: bool = False
self.interrupts: list[InterruptPayload] = []
self._lifecycle_watcher_thread: threading.Thread | None = None
+ self._lifecycle_watcher_lock = threading.Lock()
self._lifecycle_watcher_handle: SyncEventStreamHandle | None = None
self._lifecycle_cursor: int | None = None
self._lifecycle_max_reconnect_attempts = 5
@@ -1149,7 +1156,11 @@
self._transport, run_start_gate=run_start_gate
)
self._run_done = _BlockingResult()
- self._ensure_lifecycle_watcher_running()
+ # A minted thread_id names no thread until run.start creates it, and
+ # with custom auth a stream for a thread that doesn't exist yet fails,
+ # so run.start starts the watcher instead.
+ if self._explicit_thread_id:
+ self._ensure_lifecycle_watcher_running()
return self
def __exit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
@@ -1485,9 +1496,7 @@
if response is None:
return {}
if response.get("type") == "error":
- code = response.get("error", "unknown")
- message = response.get("message", "")
- raise RuntimeError(f"Protocol error [{code}]: {message}")
+ raise protocol_error(response)
meta = response.get("meta")
if isinstance(meta, dict):
applied_through_seq = meta.get("applied_through_seq")
@@ -1496,14 +1505,20 @@
return response.get("result", {})
def _ensure_lifecycle_watcher_running(self) -> None:
- if self._lifecycle_watcher_thread is not None:
- return
- self._lifecycle_watcher_thread = threading.Thread(
- target=self._run_lifecycle_watcher,
- name="langgraph-sdk-sync-lifecycle",
- daemon=True,
- )
- self._lifecycle_watcher_thread.start()
+ with self._lifecycle_watcher_lock:
+ if self._lifecycle_watcher_thread is not None:
+ return
+ if self._run_done is not None and self._run_done.done():
+ # A 4xx stopped the watcher before this run started, typically
+ # the 404 for a thread this run.start just created, so the
+ # result it left isn't this run's.
+ self._run_done = _BlockingResult()
+ self._lifecycle_watcher_thread = threading.Thread(
+ target=self._run_lifecycle_watcher,
+ name="langgraph-sdk-sync-lifecycle",
+ daemon=True,
+ )
+ self._lifecycle_watcher_thread.start()
def _observe_lifecycle_event(self, event: Event) -> None:
seq = event.get("seq")
@@ -1523,7 +1538,11 @@
if self._transport is None:
return
reconnect_attempts = 0
+ frame_error: RuntimeError | None = None
while not self._closed:
+ # A stream opened before run.start can 404 for the thread that
+ # run.start creates, so it gets one more try once a run has
started.
+ opened_before_run = not self._run_seen
try:
handle = self._transport.open_event_stream(
self._lifecycle_stream_params()
@@ -1532,28 +1551,64 @@
for event in handle.events:
if self._closed:
return
+ frame_error = stream_error_frame(event) or frame_error
self._observe_lifecycle_event(event)
self._apply_lifecycle_event(event)
err = handle.error()
if err is None:
+ # Clean EOF: stream ended without a terminal lifecycle
+ # event. Resolve `_run_done` as errored so `thread.output`
+ # doesn't block forever.
+ if not self._closed:
+ self._fail_run_done(
+ frame_error
+ or RuntimeError(
+ "lifecycle stream ended before terminal event"
+ )
+ )
return
reconnect_attempts += 1
- if reconnect_attempts > self._lifecycle_max_reconnect_attempts:
+ if (
+ is_non_retryable_stream_error(err)
+ or reconnect_attempts >
self._lifecycle_max_reconnect_attempts
+ ):
raise err
except Exception as exc:
+ if is_non_retryable_stream_error(exc):
+ if self._stop_rejected_lifecycle_watcher(
+ exc, unless_run_started=opened_before_run
+ ):
+ return
+ continue
reconnect_attempts += 1
if reconnect_attempts <=
self._lifecycle_max_reconnect_attempts:
continue
- run_done = self._run_done
- if run_done is not None and not run_done.done():
- run_done.set_result(
- _RunTerminal(
- status="errored",
- error=RuntimeError(f"Lifecycle transport failed:
{exc}"),
- )
- )
+ self._fail_run_done(
+ frame_error or RuntimeError(f"Lifecycle transport failed:
{exc}")
+ )
return
+ def _stop_rejected_lifecycle_watcher(
+ self, error: BaseException, *, unless_run_started: bool = False
+ ) -> bool:
+ """Fail the run with a 4xx `error` and clear the watcher, so run.start
starts another.
+
+ Returns False, changing nothing, when `unless_run_started` is set and a
+ run has started. Holding the lock keeps a run.start from seeing this
+ watcher as still running once it has decided to stop.
+ """
+ with self._lifecycle_watcher_lock:
+ if unless_run_started and self._run_seen:
+ return False
+ self._fail_run_done(error)
+ self._lifecycle_watcher_thread = None
+ return True
+
+ def _fail_run_done(self, error: BaseException) -> None:
+ run_done = self._run_done
+ if run_done is not None and not run_done.done():
+ run_done.set_result(_RunTerminal(status="errored", error=error))
+
def _fetch_state(self) -> dict[str, Any]:
"""Fetch the current thread state from the REST endpoint."""
return self._http.get(
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/langgraph_sdk-0.4.7/langgraph_sdk/stream/errors.py
new/langgraph_sdk-0.4.8/langgraph_sdk/stream/errors.py
--- old/langgraph_sdk-0.4.7/langgraph_sdk/stream/errors.py 1970-01-01
01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.8/langgraph_sdk/stream/errors.py 2020-02-02
01:00:00.000000000 +0100
@@ -0,0 +1,46 @@
+"""Thread-stream failures, shared by the async and sync clients."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import httpx
+from websockets.exceptions import ConnectionClosed, InvalidStatus
+
+
+def protocol_error(response: dict[str, Any]) -> RuntimeError:
+ """Build the error for a protocol `{"type": "error"}` envelope."""
+ code = response.get("error", "unknown")
+ message = response.get("message", "")
+ return RuntimeError(f"Protocol error [{code}]: {message}")
+
+
+def stream_error_frame(event: Any) -> RuntimeError | None:
+ """Return the error in an unsolicited `{"type": "error", "id": null}`
frame.
+
+ The server sends one just before it closes a thread stream whose read
failed.
+ """
+ if (
+ isinstance(event, dict)
+ and event.get("type") == "error"
+ and event.get("id") is None
+ ):
+ return protocol_error(event)
+ return None
+
+
+def is_non_retryable_stream_error(error: BaseException | None) -> bool:
+ """Whether reopening the stream would fail the same way.
+
+ True for a 4xx other than 408 and 429: the status of a refused stream
request
+ or WebSocket upgrade, or a WebSocket close code of 4000 plus that status. A
+ server that rejects a WebSocket after accepting it closes with such a code,
+ because a browser can't read the status of a refused upgrade.
+ """
+ if isinstance(error, (httpx.HTTPStatusError, InvalidStatus)):
+ status = error.response.status_code
+ elif isinstance(error, ConnectionClosed) and error.rcvd is not None:
+ status = error.rcvd.code - 4000
+ else:
+ return False
+ return 400 <= status < 500 and status not in (408, 429)
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore'
old/langgraph_sdk-0.4.7/langgraph_sdk/stream/sync_controller.py
new/langgraph_sdk-0.4.8/langgraph_sdk/stream/sync_controller.py
--- old/langgraph_sdk-0.4.7/langgraph_sdk/stream/sync_controller.py
2020-02-02 01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.8/langgraph_sdk/stream/sync_controller.py
2020-02-02 01:00:00.000000000 +0100
@@ -13,6 +13,10 @@
from langchain_protocol import Event, SubscribeParams
+from langgraph_sdk.stream.errors import (
+ is_non_retryable_stream_error,
+ stream_error_frame,
+)
from langgraph_sdk.stream.subscription import compute_union_filter,
filter_covers
from langgraph_sdk.stream.transport import (
SyncEventStreamHandle,
@@ -167,6 +171,13 @@
return
try:
for event in self._dedup_iter(shared.events):
+ if (frame_error := stream_error_frame(event)) is not None:
+ _logger.warning(
+ "Thread %s event stream failed: %s",
+ self._transport.thread_id,
+ frame_error,
+ )
+ continue
with self._lock:
if self._closed:
break
@@ -190,7 +201,11 @@
continue # rotation happened; pick up new stream
# No rotation — check if this was a transport drop
err = shared.error()
- if err is not None and not self._closed:
+ if (
+ err is not None
+ and not is_non_retryable_stream_error(err)
+ and not self._closed
+ ):
if self._reconnect_shared_stream():
continue
break
@@ -199,25 +214,6 @@
for sub in self._subscriptions.values():
sub.queue.put(None)
- def _reconnect_shared_stream(self) -> bool:
- with self._lock:
- base_filter = self._shared_stream_filter
- if base_filter is None:
- return False
- for _ in range(self._max_reconnect_attempts):
- with self._lock:
- if self._closed:
- return False
- params = self._filter_with_since(base_filter)
- try:
- new_stream = self._transport.open_event_stream(params)
- except Exception:
- continue
- with self._lock:
- self._shared_stream = new_stream
- return True
- return False
-
def _compute_current_union(
self, extra: SubscribeParams | None = None
) -> dict[str, Any]:
@@ -296,7 +292,8 @@
"""Attempt to reopen the shared stream after a transport drop.
Returns True if a new stream was successfully opened, False if all
- reconnect attempts were exhausted or the controller was closed.
+ reconnect attempts were exhausted, the server rejected the stream with
a
+ non-retryable 4xx, or the controller was closed.
"""
base_filter = self._shared_stream_filter
if base_filter is None:
@@ -317,6 +314,8 @@
old.close()
return True
except Exception as err:
+ if is_non_retryable_stream_error(err):
+ return False
_logger.debug("sync reconnect attempt %d failed: %r", attempt,
err)
return False
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/langgraph_sdk-0.4.7/tests/streaming/_events.py
new/langgraph_sdk-0.4.8/tests/streaming/_events.py
--- old/langgraph_sdk-0.4.7/tests/streaming/_events.py 2020-02-02
01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.8/tests/streaming/_events.py 2020-02-02
01:00:00.000000000 +0100
@@ -69,6 +69,16 @@
return _base(seq, "lifecycle", namespace or [], {"event": "failed",
"error": error})
+def thread_stream_failed_frame() -> dict[str, Any]:
+ """Unsolicited error frame the server sends before closing a failed
stream."""
+ return {
+ "type": "error",
+ "id": None,
+ "error": "unknown_error",
+ "message": "Thread event stream failed.",
+ }
+
+
def values_event(
seq: int = 0, namespace: list[str] | None = None, **data: Any
) -> dict[str, Any]:
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/langgraph_sdk-0.4.7/tests/streaming/_fake_server.py
new/langgraph_sdk-0.4.8/tests/streaming/_fake_server.py
--- old/langgraph_sdk-0.4.7/tests/streaming/_fake_server.py 2020-02-02
01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.8/tests/streaming/_fake_server.py 2020-02-02
01:00:00.000000000 +0100
@@ -60,6 +60,7 @@
events: list[dict[str, Any]]
delay: float = 0.0
fail_after: int | None = None
+ status: int = 200
class FakeServer:
@@ -263,6 +264,8 @@
events=list(self.scripted_events), delay=self._stream_delay
)
)
+ if script.status != 200:
+ return httpx.Response(script.status, json={"detail":
"scripted"})
self.open_event_streams += 1
self._open_event_streams_max = max(
self._open_event_streams_max, self.open_event_streams
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore'
old/langgraph_sdk-0.4.7/tests/streaming/_sync_fake_server.py
new/langgraph_sdk-0.4.8/tests/streaming/_sync_fake_server.py
--- old/langgraph_sdk-0.4.7/tests/streaming/_sync_fake_server.py
2020-02-02 01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.8/tests/streaming/_sync_fake_server.py
2020-02-02 01:00:00.000000000 +0100
@@ -14,6 +14,7 @@
class SyncStreamScript:
events: list[dict[str, Any]]
fail_after: int | None = None
+ status: int = 200
class _SseByteStream(httpx.SyncByteStream):
@@ -113,6 +114,8 @@
if self._stream_scripts
else SyncStreamScript(events=list(self.scripted_events))
)
+ if script.status != 200:
+ return httpx.Response(script.status, json={"detail":
"scripted"})
return httpx.Response(
200,
headers={"content-type": "text/event-stream"},
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore'
old/langgraph_sdk-0.4.7/tests/streaming/test_lifecycle_watcher.py
new/langgraph_sdk-0.4.8/tests/streaming/test_lifecycle_watcher.py
--- old/langgraph_sdk-0.4.7/tests/streaming/test_lifecycle_watcher.py
2020-02-02 01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.8/tests/streaming/test_lifecycle_watcher.py
2020-02-02 01:00:00.000000000 +0100
@@ -17,6 +17,7 @@
lifecycle_completed_event,
lifecycle_errored_event,
lifecycle_event,
+ thread_stream_failed_frame,
)
from streaming._fake_server import FakeServer, _StreamScript
@@ -48,7 +49,7 @@
async def test_aenter_starts_lifecycle_watcher():
- """Entering AsyncThreadStream opens lifecycle/input SSE before
run.start."""
+ """Entering with an explicit thread_id opens the lifecycle/input SSE
before run.start."""
fake = FakeServer()
fake.script([lifecycle_event(seq=0, phase="started")])
asgi = httpx.ASGITransport(app=fake.app)
@@ -274,3 +275,137 @@
assert fake.stream_request_bodies[0]["channels"] == ["lifecycle", "input"]
assert fake.stream_request_bodies[1]["channels"] == ["lifecycle", "input"]
assert fake.stream_request_bodies[1]["since"] == 1
+
+
+async def test_lifecycle_error_frame_becomes_the_run_error():
+ fake = FakeServer()
+ fake.script([lifecycle_event(seq=0, phase="running"),
thread_stream_failed_frame()])
+ async with httpx.AsyncClient(
+ transport=fake.transport, base_url="http://test";
+ ) as raw:
+ threads = ThreadsClient(HttpClient(raw))
+ async with threads.stream(thread_id="t-1", assistant_id="agent") as
thread:
+ run_done = thread._run_done
+ assert run_done is not None
+ terminal = await asyncio.wait_for(run_done, timeout=2.0)
+
+ assert terminal.status == "errored"
+ assert (
+ str(terminal.error)
+ == "Protocol error [unknown_error]: Thread event stream failed."
+ )
+
+
+async def test_lifecycle_watcher_stops_on_a_404_without_retrying():
+ fake = FakeServer()
+ fake.script_sequence([_StreamScript(events=[], status=404)])
+ async with httpx.AsyncClient(
+ transport=fake.transport, base_url="http://test";
+ ) as raw:
+ threads = ThreadsClient(HttpClient(raw))
+ async with threads.stream(thread_id="t-1", assistant_id="agent") as
thread:
+ run_done = thread._run_done
+ assert run_done is not None
+ terminal = await asyncio.wait_for(run_done, timeout=2.0)
+
+ assert isinstance(terminal.error, httpx.HTTPStatusError)
+ assert terminal.error.response.status_code == 404
+ assert len(fake.stream_request_bodies) == 1
+
+
+async def test_minted_thread_opens_the_lifecycle_stream_after_run_start():
+ fake = FakeServer()
+ fake.script([lifecycle_completed_event(seq=1)])
+ async with httpx.AsyncClient(
+ transport=fake.transport, base_url="http://test";
+ ) as raw:
+ threads = ThreadsClient(HttpClient(raw))
+ async with threads.stream(assistant_id="agent") as thread:
+ assert thread._lifecycle_watcher_task is None
+ await thread.run.start(input={})
+ run_done = thread._run_done
+ assert run_done is not None
+ terminal = await asyncio.wait_for(run_done, timeout=2.0)
+
+ assert terminal.status == "completed"
+
+
+def _new_thread_scripts() -> list[_StreamScript]:
+ return [
+ _StreamScript(events=[], status=404),
+ _StreamScript(events=[lifecycle_completed_event(seq=1)]),
+ ]
+
+
+async def
test_run_start_restarts_a_watcher_stopped_by_a_404_for_a_new_thread():
+ fake = FakeServer()
+ fake.script_sequence(_new_thread_scripts())
+ fake.set_state({"ok": True})
+ async with httpx.AsyncClient(
+ transport=fake.transport, base_url="http://test";
+ ) as raw:
+ threads = ThreadsClient(HttpClient(raw))
+ async with threads.stream(thread_id="t-new", assistant_id="agent") as
thread:
+ rejected = thread._run_done
+ assert rejected is not None
+ await asyncio.wait_for(rejected, timeout=2.0)
+ await thread.run.start(input={})
+ output = await thread.output.with_timeout(2.0)
+
+ assert output == {"ok": True}
+ assert len(fake.stream_request_bodies) == 2
+
+
+async def test_lifecycle_watcher_retries_a_404_that_arrives_after_run_start():
+ fake = FakeServer()
+ fake.script_sequence(_new_thread_scripts())
+ fake.set_state({"ok": True})
+ stream_opened = asyncio.Event()
+ run_started = asyncio.Event()
+
+ async def handle(request: httpx.Request) -> httpx.Response:
+ if (
+ request.url.path.endswith("/stream/events")
+ and not fake.stream_request_bodies
+ ):
+ stream_opened.set()
+ await run_started.wait()
+ return await fake._handle_request(request)
+
+ async with httpx.AsyncClient(
+ transport=httpx.MockTransport(handle), base_url="http://test";
+ ) as raw:
+ threads = ThreadsClient(HttpClient(raw))
+ async with threads.stream(thread_id="t-new", assistant_id="agent") as
thread:
+ await asyncio.wait_for(stream_opened.wait(), timeout=2.0)
+ await thread.run.start(input={})
+ run_started.set()
+ output = await thread.output.with_timeout(2.0)
+
+ assert output == {"ok": True}
+ assert len(fake.stream_request_bodies) == 2
+
+
+async def
test_lifecycle_watcher_reports_the_error_frame_when_reconnects_run_out():
+ fake = FakeServer()
+ async with httpx.AsyncClient(
+ transport=fake.transport, base_url="http://test";
+ ) as raw:
+ threads = ThreadsClient(HttpClient(raw))
+ stream = threads.stream(thread_id="t-1", assistant_id="agent")
+ stream._lifecycle_max_reconnect_attempts = 1
+ fake.script_sequence(
+ [
+ _StreamScript(events=[thread_stream_failed_frame()],
fail_after=1)
+ for _ in range(stream._lifecycle_max_reconnect_attempts + 1)
+ ]
+ )
+ async with stream as thread:
+ run_done = thread._run_done
+ assert run_done is not None
+ terminal = await asyncio.wait_for(run_done, timeout=2.0)
+
+ assert (
+ str(terminal.error)
+ == "Protocol error [unknown_error]: Thread event stream failed."
+ )
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore'
old/langgraph_sdk-0.4.7/tests/streaming/test_shared_stream.py
new/langgraph_sdk-0.4.8/tests/streaming/test_shared_stream.py
--- old/langgraph_sdk-0.4.7/tests/streaming/test_shared_stream.py
2020-02-02 01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.8/tests/streaming/test_shared_stream.py
2020-02-02 01:00:00.000000000 +0100
@@ -1,11 +1,13 @@
from __future__ import annotations
import asyncio
-from collections.abc import AsyncGenerator
+from collections.abc import AsyncGenerator, AsyncIterator
from typing import Any, cast
from unittest.mock import MagicMock
import httpx
+from websockets.exceptions import ConnectionClosedError
+from websockets.frames import Close
from langgraph_sdk._async.http import HttpClient
from langgraph_sdk._async.threads import ThreadsClient
@@ -348,3 +350,66 @@
]
assert len(values_requests) == 1
assert values_requests[0]["since"] == 17
+
+
+async def _drain(events: AsyncIterator[Any]) -> list[Any]:
+ return [event async for event in events]
+
+
+async def test_shared_stream_stops_reconnecting_on_a_404():
+ fake = FakeServer()
+ fake.script_sequence(
+ [
+ _StreamScript(events=[values_event(seq=1)], fail_after=1),
+ _StreamScript(events=[], status=404),
+ ]
+ )
+ async with httpx.AsyncClient(
+ transport=fake.transport, base_url="http://test";
+ ) as raw:
+ threads = ThreadsClient(HttpClient(raw))
+ async with threads.stream(assistant_id="agent") as thread:
+ received = await asyncio.wait_for(
+ _drain(thread.subscribe(["values"])), timeout=5.0
+ )
+
+ assert [event["seq"] for event in received] == [1]
+ assert len(fake.stream_request_bodies) == 2
+
+
+async def test_shared_stream_does_not_reconnect_after_a_4404_close(
+ monkeypatch: Any,
+) -> None:
+ opened: list[dict[str, Any]] = []
+
+ def open_then_close_with_4404(
+ _self: ProtocolSseTransport, params: dict[str, Any]
+ ) -> EventStreamHandle:
+ opened.append(params)
+ loop = asyncio.get_running_loop()
+ ready: asyncio.Future[None] = loop.create_future()
+ ready.set_result(None)
+ done: asyncio.Future[BaseException | None] = loop.create_future()
+ done.set_result(ConnectionClosedError(Close(4404, "not found"), None))
+
+ async def events() -> AsyncGenerator[Any, None]:
+ await asyncio.sleep(0)
+ yield values_event(seq=1)
+
+ async def close() -> None:
+ return None
+
+ return EventStreamHandle(events=events(), ready=ready, done=done,
close=close)
+
+ monkeypatch.setattr(
+ ProtocolSseTransport, "open_event_stream", open_then_close_with_4404
+ )
+ async with httpx.AsyncClient(base_url="http://test";) as raw:
+ threads = ThreadsClient(HttpClient(raw))
+ async with threads.stream(assistant_id="agent") as thread:
+ received = await asyncio.wait_for(
+ _drain(thread.subscribe(["values"])), timeout=5.0
+ )
+
+ assert [event["seq"] for event in received] == [1]
+ assert len(opened) == 1
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore'
old/langgraph_sdk-0.4.7/tests/streaming/test_stream_errors.py
new/langgraph_sdk-0.4.8/tests/streaming/test_stream_errors.py
--- old/langgraph_sdk-0.4.7/tests/streaming/test_stream_errors.py
1970-01-01 01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.8/tests/streaming/test_stream_errors.py
2020-02-02 01:00:00.000000000 +0100
@@ -0,0 +1,67 @@
+from __future__ import annotations
+
+import httpx
+import pytest
+from websockets.datastructures import Headers
+from websockets.exceptions import ConnectionClosedError, InvalidStatus
+from websockets.frames import Close
+from websockets.http11 import Response
+
+from langgraph_sdk.stream.errors import (
+ is_non_retryable_stream_error,
+ stream_error_frame,
+)
+
+
+def _http_error(status: int) -> httpx.HTTPStatusError:
+ request = httpx.Request("POST", "http://test/threads/t-1/stream/events";)
+ response = httpx.Response(status, request=request)
+ return httpx.HTTPStatusError("scripted", request=request,
response=response)
+
+
+def _ws_close(code: int) -> ConnectionClosedError:
+ return ConnectionClosedError(Close(code, "scripted"), None)
+
+
[email protected](
+ ("error", "expected"),
+ [
+ pytest.param(_http_error(403), True, id="http-403"),
+ pytest.param(_http_error(404), True, id="http-404"),
+ pytest.param(_http_error(408), False, id="http-408"),
+ pytest.param(_http_error(429), False, id="http-429"),
+ pytest.param(_http_error(503), False, id="http-503"),
+ pytest.param(_ws_close(4404), True, id="ws-close-4404"),
+ pytest.param(_ws_close(4422), True, id="ws-close-4422"),
+ pytest.param(_ws_close(4429), False, id="ws-close-4429"),
+ pytest.param(_ws_close(4503), False, id="ws-close-4503"),
+ pytest.param(_ws_close(1011), False, id="ws-close-1011"),
+ pytest.param(
+ InvalidStatus(Response(403, "Forbidden", Headers())),
+ True,
+ id="ws-upgrade-403",
+ ),
+ pytest.param(RuntimeError("network down"), False, id="no-status"),
+ pytest.param(None, False, id="none"),
+ ],
+)
+def test_is_non_retryable_stream_error(
+ error: BaseException | None, expected: bool
+) -> None:
+ assert is_non_retryable_stream_error(error) is expected
+
+
+def test_stream_error_frame_reads_only_unsolicited_error_frames() -> None:
+ frame = {
+ "type": "error",
+ "id": None,
+ "error": "unknown_error",
+ "message": "Thread event stream failed.",
+ }
+
+ assert (
+ str(stream_error_frame(frame))
+ == "Protocol error [unknown_error]: Thread event stream failed."
+ )
+ assert stream_error_frame({**frame, "id": 3}) is None
+ assert stream_error_frame({"type": "event", "method": "values"}) is None
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore'
old/langgraph_sdk-0.4.7/tests/streaming/test_sync_thread_stream.py
new/langgraph_sdk-0.4.8/tests/streaming/test_sync_thread_stream.py
--- old/langgraph_sdk-0.4.7/tests/streaming/test_sync_thread_stream.py
2020-02-02 01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.8/tests/streaming/test_sync_thread_stream.py
2020-02-02 01:00:00.000000000 +0100
@@ -35,6 +35,7 @@
message_text_delta_event,
message_text_finish_event,
tasks_start_event,
+ thread_stream_failed_frame,
tool_finished_event,
tool_started_event,
updates_event,
@@ -868,3 +869,196 @@
]
assert {"scope": "root"} in checkpoints
assert {"scope": "child"} not in checkpoints
+
+
+def test_sync_lifecycle_clean_eof_fails_the_run_instead_of_blocking():
+ fake = SyncFakeServer()
+ fake.script([lifecycle_started_event(seq=0)])
+ 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:
+ run_done = thread._run_done
+ assert run_done is not None
+ terminal = run_done.result(timeout=2.0)
+
+ assert terminal.status == "errored"
+ assert "ended before terminal" in str(terminal.error)
+
+
+def test_sync_lifecycle_error_frame_becomes_the_run_error():
+ fake = SyncFakeServer()
+ fake.script([lifecycle_event(seq=0, phase="running"),
thread_stream_failed_frame()])
+ 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:
+ run_done = thread._run_done
+ assert run_done is not None
+ terminal = run_done.result(timeout=2.0)
+
+ assert terminal.status == "errored"
+ assert (
+ str(terminal.error)
+ == "Protocol error [unknown_error]: Thread event stream failed."
+ )
+
+
+def test_sync_lifecycle_watcher_stops_on_a_404_without_retrying():
+ fake = SyncFakeServer()
+ fake.script_sequence([SyncStreamScript(events=[], status=404)])
+ 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:
+ run_done = thread._run_done
+ assert run_done is not None
+ terminal = run_done.result(timeout=2.0)
+
+ assert isinstance(terminal.error, httpx.HTTPStatusError)
+ assert terminal.error.response.status_code == 404
+ assert len(fake.stream_request_bodies) == 1
+
+
+def test_sync_minted_thread_opens_the_lifecycle_stream_after_run_start():
+ fake = SyncFakeServer()
+ fake.script([lifecycle_completed_event(seq=1)])
+ with httpx.Client(transport=fake.transport, base_url="http://test";) as raw:
+ threads = SyncThreadsClient(SyncHttpClient(raw))
+ with threads.stream(assistant_id="agent") as thread:
+ assert thread._lifecycle_watcher_thread is None
+ thread.run.start(input={})
+ run_done = thread._run_done
+ assert run_done is not None
+ terminal = run_done.result(timeout=2.0)
+
+ assert terminal.status == "completed"
+
+
+def _sync_new_thread_scripts() -> list[SyncStreamScript]:
+ return [
+ SyncStreamScript(events=[], status=404),
+ SyncStreamScript(events=[lifecycle_completed_event(seq=1)]),
+ ]
+
+
+def test_sync_run_start_restarts_a_watcher_stopped_by_a_404_for_a_new_thread():
+ fake = SyncFakeServer()
+ fake.script_sequence(_sync_new_thread_scripts())
+ fake.set_state({"ok": True})
+ with httpx.Client(transport=fake.transport, base_url="http://test";) as raw:
+ threads = SyncThreadsClient(SyncHttpClient(raw))
+ with threads.stream(thread_id="t-new", assistant_id="agent") as thread:
+ rejected = thread._run_done
+ assert rejected is not None
+ rejected.result(timeout=2.0)
+ thread.run.start(input={})
+ output = thread.output
+
+ assert output == {"ok": True}
+ assert len(fake.stream_request_bodies) == 2
+
+
+def test_sync_lifecycle_watcher_retries_a_404_that_arrives_after_run_start():
+ fake = SyncFakeServer()
+ fake.script_sequence(_sync_new_thread_scripts())
+ fake.set_state({"ok": True})
+ stream_opened = threading.Event()
+ run_started = threading.Event()
+
+ def handle(request: httpx.Request) -> httpx.Response:
+ if (
+ request.url.path.endswith("/stream/events")
+ and not fake.stream_request_bodies
+ ):
+ stream_opened.set()
+ run_started.wait(timeout=2.0)
+ return fake._handle(request)
+
+ with httpx.Client(
+ transport=httpx.MockTransport(handle), base_url="http://test";
+ ) as raw:
+ threads = SyncThreadsClient(SyncHttpClient(raw))
+ with threads.stream(thread_id="t-new", assistant_id="agent") as thread:
+ assert stream_opened.wait(timeout=2.0)
+ thread.run.start(input={})
+ run_started.set()
+ output = thread.output
+
+ assert output == {"ok": True}
+ assert len(fake.stream_request_bodies) == 2
+
+
+def
test_sync_lifecycle_watcher_reports_the_error_frame_when_reconnects_run_out():
+ fake = SyncFakeServer()
+ with httpx.Client(transport=fake.transport, base_url="http://test";) as raw:
+ threads = SyncThreadsClient(SyncHttpClient(raw))
+ stream = threads.stream(thread_id="t-1", assistant_id="agent")
+ stream._lifecycle_max_reconnect_attempts = 1
+ fake.script_sequence(
+ [
+ SyncStreamScript(events=[thread_stream_failed_frame()],
fail_after=1)
+ for _ in range(stream._lifecycle_max_reconnect_attempts + 1)
+ ]
+ )
+ with stream as thread:
+ run_done = thread._run_done
+ assert run_done is not None
+ terminal = run_done.result(timeout=2.0)
+
+ assert (
+ str(terminal.error)
+ == "Protocol error [unknown_error]: Thread event stream failed."
+ )
+
+
+def test_sync_concurrent_run_starts_start_one_lifecycle_watcher(monkeypatch):
+ real_thread = threading.Thread
+ created_watchers: list[str] = []
+
+ class _SlowToCreateThread(real_thread):
+ def __init__(self, *args: Any, **kwargs: Any) -> None:
+ if kwargs.get("name") == "langgraph-sdk-sync-lifecycle":
+ created_watchers.append(kwargs["name"])
+ time.sleep(0.05)
+ super().__init__(*args, **kwargs)
+
+ monkeypatch.setattr(threading, "Thread", _SlowToCreateThread)
+ fake = SyncFakeServer()
+ fake.script([lifecycle_completed_event(seq=1)])
+ with httpx.Client(transport=fake.transport, base_url="http://test";) as raw:
+ threads = SyncThreadsClient(SyncHttpClient(raw))
+ with threads.stream(assistant_id="agent") as thread:
+ barrier = threading.Barrier(2)
+
+ def start_run() -> None:
+ barrier.wait()
+ thread.run.start(input={})
+
+ runners = [real_thread(target=start_run) for _ in range(2)]
+ for runner in runners:
+ runner.start()
+ for runner in runners:
+ runner.join()
+
+ assert len(created_watchers) == 1
+
+
+def test_sync_reconnect_stops_on_a_404(monkeypatch):
+ monkeypatch.setattr(_ctrl_mod.time, "sleep", lambda _delay: None)
+ opened: list[dict[str, Any]] = []
+
+ class _NotFoundTransport(SyncProtocolSseTransport):
+ def open_event_stream(self, params: dict) -> SyncEventStreamHandle:
+ opened.append(params)
+ request = httpx.Request("POST",
"http://test/threads/t-1/stream/events";)
+ raise httpx.HTTPStatusError(
+ "not found",
+ request=request,
+ response=httpx.Response(404, request=request),
+ )
+
+ with httpx.Client(base_url="http://test";) as raw:
+ transport = _NotFoundTransport(client=raw, thread_id="t-1")
+ controller = SyncStreamController(transport, max_reconnect_attempts=5)
+ controller._shared_stream_filter = {"channels": ["values"]}
+ assert controller._reconnect_shared_stream() is False
+
+ assert len(opened) == 1
diff -urN '--exclude=CVS' '--exclude=.cvsignore' '--exclude=.svn'
'--exclude=.svnignore' old/langgraph_sdk-0.4.7/uv.lock
new/langgraph_sdk-0.4.8/uv.lock
--- old/langgraph_sdk-0.4.7/uv.lock 2020-02-02 01:00:00.000000000 +0100
+++ new/langgraph_sdk-0.4.8/uv.lock 2020-02-02 01:00:00.000000000 +0100
@@ -301,7 +301,7 @@
[[package]]
name = "langgraph"
-version = "1.2.15"
+version = "1.2.16"
source = { editable = "../langgraph" }
dependencies = [
{ name = "langchain-core" },
@@ -433,7 +433,7 @@
[[package]]
name = "langgraph-prebuilt"
-version = "1.1.0"
+version = "1.1.1"
source = { editable = "../prebuilt" }
dependencies = [
{ name = "langchain-core" },
