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 == []