WEBDIENSTE/langgraph
1
0
Fork 0
mirror of https://github.com/langchain-ai/langgraph.git synced 2026-10-21 15:31:43 +02:00
Code Issues Packages Projects Releases Wiki Activity

fix(sdk-py): surface thread stream errors and stop retrying a rejected stream (#9317)

Browse Source 
## Summary

Thread streams in `langgraph_sdk` now report why they ended. An `id:
null` error frame becomes the error `thread.output` raises, the sync
client no longer blocks forever when the server closes the stream, and a
stream the server rejects with a 4xx ends at once instead of being
retried.

## Problem

- A server sends `{"type": "error", "id": null, ...}` just before it
closes a thread stream whose read failed. The SDK dropped the frame, so
`thread.output` raised a generic "lifecycle stream ended before terminal
event" (async) or blocked forever (sync: the lifecycle watcher returned
without settling the run).
- A 403 or 404 on the stream request was retried like a network error,
then reported as `RuntimeError("Lifecycle transport failed: ...")` with
no status. A server that rejects a WebSocket after accepting it closes
with 4000 plus the HTTP status (4403, 4404, 4422), since a browser can't
read a refused upgrade's status. The shared stream counted each such
reconnect as a success and reconnected forever.

## Fix

- The lifecycle watcher fails the run with the frame's error on a clean
close (both clients). The shared stream logs the frame instead of
dropping it: projections still end when the stream closes, as before,
and `thread.output` carries the error.
- A 4xx other than 408 and 429, as a status or as a 4000+status close
code, ends the stream without retries. `thread.output` now raises that
`httpx.HTTPStatusError` or `websockets.exceptions.ConnectionClosedError`
instead of the generic `RuntimeError`.
- For a minted `thread_id`, the lifecycle watcher starts after
`run.start` creates the thread. Without this, the 4xx stop would turn a
stream opened under custom auth for a not-yet-created thread from
retried into failed. The sync client starts it under a lock, since
`run.start` can be called from several threads.
- Drops the first of two `SyncStreamController._reconnect_shared_stream`
definitions. It was shadowed by the second and never ran.

Not in this PR: over WebSocket the server closes with 1011 after an
error frame, which is still retried, while over SSE the close is clean
and fails the stream. Reconnecting after a clean close, as
`langchain-ai/langgraphjs#2812` does for the JS SDK, is a separate
change.

## Test plan

- [x] The 10 new behavior tests (async and sync: error frame, clean
close, 404, 4404 close, reconnect on 404, minted `thread_id`) fail on
`main` and pass here; a race test checks that two concurrent `run.start`
calls start one watcher.
- [x] `make test` in `libs/sdk-py`: 541 passed. `make lint` clean,
`check_sdk_methods.py` passes.
- [x] `RemoteGraph` v3 tests in `libs/langgraph`
(`test_remote_graph_v3.py`,
`test_stream_events_v3_kwarg_forwarding.py`): 94 passed against this
SDK.
- [x] Against a local server with custom auth: a thread the caller can't
read fails at once with 404 (SSE) or 4404 (WebSocket), and a new thread
works with a 1 s pause before `run.start`, which fails on `main`.
This commit is contained in:
Elior Nataf Lackritz
2026-10-16 15:29:34 -04:00
committed by GitHub
parent b0cabd93d0
commit 9e6fc45c6f
11 changed files with 686 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
libs/sdk-py/langgraph_sdk/_async/stream.py
+65 -26
View File
@@ -15,6 +15,7 @@ from __future__ import annotations
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 @@ from langgraph_sdk.stream.decoders import (
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 @@ from langgraph_sdk.stream.transport import (
ProtocolWebSocketTransport,
)
logger = logging.getLogger(__name__)
class InterruptPayload(TypedDict):
"""Payload surfaced when the server requests human input for a thread."""
@@ -193,6 +201,7 @@ class RunModule:
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 @@ class AsyncThreadStream:
# 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 @@ class AsyncThreadStream:
# 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 @@ class AsyncThreadStream:
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 @@ class AsyncThreadStream:
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 @@ class AsyncThreadStream:
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 @@ class AsyncThreadStream:
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 @@ class AsyncThreadStream:
# 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 @@ class AsyncThreadStream:
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 @@ class AsyncThreadStream:
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 @@ class AsyncThreadStream:
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 @@ class AsyncThreadStream:
# 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(
libs/sdk-py/langgraph_sdk/_sync/stream.py
+76 -21
View File
@@ -34,6 +34,11 @@ from langgraph_sdk.stream.decoders import (
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 @@ class SyncRunModule:
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 @@ class SyncThreadStream:
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 @@ class SyncThreadStream:
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 @@ class SyncThreadStream:
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 @@ class SyncThreadStream:
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 @@ class SyncThreadStream:
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 @@ class SyncThreadStream:
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(
libs/sdk-py/langgraph_sdk/stream/errors.py
+46
View File
@@ -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)
libs/sdk-py/langgraph_sdk/stream/sync_controller.py
+20 -21
View File
@@ -13,6 +13,10 @@ from typing import Any
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 @@ class SyncStreamController:
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 @@ class SyncStreamController:
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 @@ class SyncStreamController:
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 @@ class SyncStreamController:
"""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 @@ class SyncStreamController:
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
libs/sdk-py/tests/streaming/_events.py
+10
View File
@@ -69,6 +69,16 @@ def lifecycle_errored_event(
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]:
libs/sdk-py/tests/streaming/_fake_server.py
+3
View File
@@ -60,6 +60,7 @@ class _StreamScript:
events: list[dict[str, Any]]
delay: float = 0.0
fail_after: int | None = None
status: int = 200
class FakeServer:
@@ -263,6 +264,8 @@ class FakeServer:
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
libs/sdk-py/tests/streaming/_sync_fake_server.py
+3
View File
@@ -14,6 +14,7 @@ import orjson
class SyncStreamScript:
events: list[dict[str, Any]]
fail_after: int | None = None
status: int = 200
class _SseByteStream(httpx.SyncByteStream):
@@ -113,6 +114,8 @@ class SyncFakeServer:
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"},
libs/sdk-py/tests/streaming/test_lifecycle_watcher.py
+136 -1
View File
@@ -17,6 +17,7 @@ from streaming._events import (
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_interrupts_populated_from_input_requested_event():
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 @@ async def test_lifecycle_watcher_reconnects_with_since_after_transport_drop():
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."
)
libs/sdk-py/tests/streaming/test_shared_stream.py
+66 -1
View File
@@ -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 @@ async def test_send_command_applied_through_seq_seeds_shared_stream_since():
]
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
libs/sdk-py/tests/streaming/test_stream_errors.py
+67
View File
@@ -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)
@pytest.mark.parametrize(
("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
libs/sdk-py/tests/streaming/test_sync_thread_stream.py
+194
View File
@@ -35,6 +35,7 @@ from streaming._events import (
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 @@ def test_interleave_projections_data_channel_scoped_to_root_namespace():
]
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