From e901fc1746183f77d4d90579c37136a13fb2806d Mon Sep 17 00:00:00 2001 From: Elior Nataf Lackritz Date: Fri, 16 Oct 2026 13:37:35 -0400 Subject: [PATCH] fix(sdk-py): restart the lifecycle watcher when run.start creates the thread An explicit thread_id can name a thread that run.start is about to create. Under custom auth its lifecycle stream gets a 404 first, which stopped the watcher for good, so thread.output raised that 404 after a successful run. run.start now starts a fresh watcher and run result when the watcher has stopped, and a 404 on a stream opened before run.start gets one more try once a run has started. Also reports the last error frame when lifecycle reconnects run out, and checks the stream error types with isinstance instead of getattr. --- libs/sdk-py/langgraph_sdk/_async/stream.py | 19 ++++- libs/sdk-py/langgraph_sdk/_sync/stream.py | 39 +++++++-- libs/sdk-py/langgraph_sdk/stream/errors.py | 24 +++--- .../tests/streaming/test_lifecycle_watcher.py | 81 +++++++++++++++++++ .../streaming/test_sync_thread_stream.py | 77 ++++++++++++++++++ 5 files changed, 222 insertions(+), 18 deletions(-) diff --git a/libs/sdk-py/langgraph_sdk/_async/stream.py b/libs/sdk-py/langgraph_sdk/_async/stream.py index eaa9dac19..f4c49b773 100644 --- a/libs/sdk-py/langgraph_sdk/_async/stream.py +++ b/libs/sdk-py/langgraph_sdk/_async/stream.py @@ -1847,8 +1847,14 @@ 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 task.done(): return + if task is not None: + # The watcher stopped before this run started, typically on 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_task = asyncio.create_task( self._run_lifecycle_watcher() ) @@ -1871,8 +1877,11 @@ class AsyncThreadStream: if self._transport is None: return reconnect_attempts = 0 + frame_error: RuntimeError | None = None while not self._closed: - frame_error: RuntimeError | None = None + # 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() @@ -1909,13 +1918,17 @@ class AsyncThreadStream: raise except Exception as exc: if is_non_retryable_stream_error(exc): + if opened_before_run and self._run_seen: + continue self._fail_run_done(exc) return reconnect_attempts += 1 if reconnect_attempts <= self._lifecycle_max_reconnect_attempts: await asyncio.sleep(0.05) continue - self._fail_run_done(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: diff --git a/libs/sdk-py/langgraph_sdk/_sync/stream.py b/libs/sdk-py/langgraph_sdk/_sync/stream.py index 89ff12f88..d4fdaac9f 100644 --- a/libs/sdk-py/langgraph_sdk/_sync/stream.py +++ b/libs/sdk-py/langgraph_sdk/_sync/stream.py @@ -1508,6 +1508,11 @@ class SyncThreadStream: 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(): + # The watcher stopped before this run started, typically on 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", @@ -1533,8 +1538,11 @@ class SyncThreadStream: if self._transport is None: return reconnect_attempts = 0 + frame_error: RuntimeError | None = None while not self._closed: - frame_error: RuntimeError | None = None + # 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() @@ -1552,7 +1560,7 @@ class SyncThreadStream: # event. Resolve `_run_done` as errored so `thread.output` # doesn't block forever. if not self._closed: - self._fail_run_done( + self._stop_lifecycle_watcher( frame_error or RuntimeError( "lifecycle stream ended before terminal event" @@ -1567,14 +1575,35 @@ class SyncThreadStream: raise err except Exception as exc: if is_non_retryable_stream_error(exc): - self._fail_run_done(exc) - return + if self._stop_lifecycle_watcher( + exc, unless_run_started=opened_before_run + ): + return + continue reconnect_attempts += 1 if reconnect_attempts <= self._lifecycle_max_reconnect_attempts: continue - self._fail_run_done(RuntimeError(f"Lifecycle transport failed: {exc}")) + self._stop_lifecycle_watcher( + frame_error or RuntimeError(f"Lifecycle transport failed: {exc}") + ) return + def _stop_lifecycle_watcher( + self, error: BaseException, *, unless_run_started: bool = False + ) -> bool: + """Fail the run with `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(): diff --git a/libs/sdk-py/langgraph_sdk/stream/errors.py b/libs/sdk-py/langgraph_sdk/stream/errors.py index 5f9f2d90c..2c36ae3c7 100644 --- a/libs/sdk-py/langgraph_sdk/stream/errors.py +++ b/libs/sdk-py/langgraph_sdk/stream/errors.py @@ -4,6 +4,9 @@ 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.""" @@ -29,14 +32,15 @@ def stream_error_frame(event: Any) -> RuntimeError | 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, whether it is the status of the stream - request 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. + 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. """ - status = getattr(getattr(error, "response", None), "status_code", None) - if status is None: - code = getattr(getattr(error, "rcvd", None), "code", None) - if isinstance(code, int) and 4000 <= code < 5000: - status = code - 4000 - return isinstance(status, int) and 400 <= status < 500 and status not in (408, 429) + 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 --git a/libs/sdk-py/tests/streaming/test_lifecycle_watcher.py b/libs/sdk-py/tests/streaming/test_lifecycle_watcher.py index da1eb1eca..c68264116 100644 --- a/libs/sdk-py/tests/streaming/test_lifecycle_watcher.py +++ b/libs/sdk-py/tests/streaming/test_lifecycle_watcher.py @@ -328,3 +328,84 @@ async def test_minted_thread_opens_the_lifecycle_stream_after_run_start(): 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 --git a/libs/sdk-py/tests/streaming/test_sync_thread_stream.py b/libs/sdk-py/tests/streaming/test_sync_thread_stream.py index 67731497c..651b5d3c6 100644 --- a/libs/sdk-py/tests/streaming/test_sync_thread_stream.py +++ b/libs/sdk-py/tests/streaming/test_sync_thread_stream.py @@ -932,6 +932,83 @@ def test_sync_minted_thread_opens_the_lifecycle_stream_after_run_start(): 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] = []