From 89fad2ab8bfb0cf45eda0920394f1fd620971dfc Mon Sep 17 00:00:00 2001 From: John Kennedy Date: Sat, 17 Oct 2026 01:44:27 +0000 Subject: [PATCH] fix: isolate test server and wait for root lifecycle completion Run the locked dev server on supported Python separately from the core test matrix. Ignore child terminal lifecycle events until the root finishes committing state. Co-authored-by: open-swe[bot] --- libs/langgraph/Makefile | 8 ++- libs/sdk-py/langgraph_sdk/_async/stream.py | 4 ++ libs/sdk-py/langgraph_sdk/_sync/stream.py | 4 ++ .../streaming/test_lifecycle_namespace.py | 67 +++++++++++++++++++ 4 files changed, 82 insertions(+), 1 deletion(-) create mode 100644 libs/sdk-py/tests/streaming/test_lifecycle_namespace.py diff --git a/libs/langgraph/Makefile b/libs/langgraph/Makefile index c5ce24a43..1ce5f4f6e 100644 --- a/libs/langgraph/Makefile +++ b/libs/langgraph/Makefile @@ -43,9 +43,15 @@ start-services: stop-services: docker compose -f tests/compose-postgres.yml -f tests/compose-redis.yml down -v +# The test matrix includes 3.10 and 3.14, but the locked in-memory server +# dependencies are available on 3.11-3.13. Keep its interpreter separate. +DEV_SERVER_PYTHON ?= 3.12 +DEV_SERVER_ENV = $(CURDIR)/.venvs/dev-server + start-dev-server: uv sync --frozen --group test --no-dev - @LOG_LEVEL=warning uv run --no-sync langgraph dev --config tests/example_app/langgraph.json --no-browser --no-reload & \ + UV_PROJECT_ENVIRONMENT=$(DEV_SERVER_ENV) uv sync --frozen --group test --no-dev --python $(DEV_SERVER_PYTHON) + @LOG_LEVEL=warning UV_PROJECT_ENVIRONMENT=$(DEV_SERVER_ENV) uv run --no-sync --python $(DEV_SERVER_PYTHON) langgraph dev --config tests/example_app/langgraph.json --no-browser --no-reload & \ PID=$$!; echo "$$PID" > .devserver.pid; \ for ATTEMPT in $$(seq 1 60); do \ if ! kill -0 $$PID 2>/dev/null; then break; fi; \ diff --git a/libs/sdk-py/langgraph_sdk/_async/stream.py b/libs/sdk-py/langgraph_sdk/_async/stream.py index bc7e68ff6..790a2fdcd 100644 --- a/libs/sdk-py/langgraph_sdk/_async/stream.py +++ b/libs/sdk-py/langgraph_sdk/_async/stream.py @@ -1972,6 +1972,10 @@ class AsyncThreadStream: # knows a run exists (handles reattach without run.start). self._run_seen = True elif phase in ("completed", "failed"): + # A child may finish before its parent commits the child's update. + # Only the root lifecycle can make thread.output terminal. + if isinstance(params, dict) and params.get("namespace"): + return # Why: interrupts describe current-run state. Clear on terminal # lifecycle so a subsequent run.respond() can't fire against a # stale prior-run interrupt_id. Acquire `_interrupts_lock` so diff --git a/libs/sdk-py/langgraph_sdk/_sync/stream.py b/libs/sdk-py/langgraph_sdk/_sync/stream.py index fabbff24d..3c3e0ce23 100644 --- a/libs/sdk-py/langgraph_sdk/_sync/stream.py +++ b/libs/sdk-py/langgraph_sdk/_sync/stream.py @@ -1614,6 +1614,10 @@ class SyncThreadStream: if phase in ("started", "running"): self._run_seen = True elif phase in ("completed", "failed"): + # A child may finish before its parent commits the child's update. + # Only the root lifecycle can make thread.output terminal. + if isinstance(params, dict) and params.get("namespace"): + return self.interrupted = False self.interrupts = [] run_done = self._run_done diff --git a/libs/sdk-py/tests/streaming/test_lifecycle_namespace.py b/libs/sdk-py/tests/streaming/test_lifecycle_namespace.py new file mode 100644 index 000000000..e22bb51f6 --- /dev/null +++ b/libs/sdk-py/tests/streaming/test_lifecycle_namespace.py @@ -0,0 +1,67 @@ +"""Child lifecycle completion must not finish the parent thread's output.""" + +from __future__ import annotations + +import asyncio +from typing import cast + +import httpx +import pytest +from langchain_protocol import Event + +from langgraph_sdk._async.http import HttpClient +from langgraph_sdk._async.threads import ThreadsClient +from langgraph_sdk._sync.http import SyncHttpClient +from langgraph_sdk._sync.stream import _BlockingResult +from langgraph_sdk._sync.threads import SyncThreadsClient +from streaming._events import input_requested_event, lifecycle_event + + +@pytest.mark.parametrize("phase", ["completed", "failed"]) +async def test_async_child_terminal_does_not_finish_root(phase): + async with httpx.AsyncClient(base_url="http://test") as raw: + thread = ThreadsClient(HttpClient(raw)).stream( + thread_id="t-1", assistant_id="agent" + ) + thread._run_done = asyncio.get_running_loop().create_future() + await thread._apply_lifecycle_event( + cast(Event, input_requested_event(namespace=["sub:1"])) + ) + await thread._apply_lifecycle_event( + cast(Event, lifecycle_event(namespace=["sub:1"], event=phase)) + ) + assert not thread._run_done.done() + assert thread.interrupted + assert thread.interrupts[0]["namespace"] == ["sub:1"] + + await thread._apply_lifecycle_event( + cast(Event, lifecycle_event(event="completed")) + ) + assert thread._run_done.done() + assert (await thread._run_done).error is None + assert not thread.interrupted + assert thread.interrupts == [] + + +@pytest.mark.parametrize("phase", ["completed", "failed"]) +def test_sync_child_terminal_does_not_finish_root(phase): + with httpx.Client(base_url="http://test") as raw: + thread = SyncThreadsClient(SyncHttpClient(raw)).stream( + thread_id="t-1", assistant_id="agent" + ) + thread._run_done = _BlockingResult() + thread._apply_lifecycle_event( + cast(Event, input_requested_event(namespace=["sub:1"])) + ) + thread._apply_lifecycle_event( + cast(Event, lifecycle_event(namespace=["sub:1"], event=phase)) + ) + assert not thread._run_done.done() + assert thread.interrupted + assert thread.interrupts[0]["namespace"] == ["sub:1"] + + thread._apply_lifecycle_event(cast(Event, lifecycle_event(event="completed"))) + assert thread._run_done.done() + assert thread._run_done.result().error is None + assert not thread.interrupted + assert thread.interrupts == []